int global_seqno = 0; //#define RANDOM_DROP 1 int my_token_held = 0; int my_do_delivery = 0; unsigned long long token_ring_id_seq = 0; int log_digest = 0; int last_released = 0; int set_aru = -1; int totemsrp_brake; /* * Copyright (c) 2003-2004 MontaVista Software, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@mvista.com) * * This software licensed under BSD license, the text of which follows: * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * - Redistributions of source code must retain the above copyright notice, * this list of conditions and the following disclaimer. * - Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * - Neither the name of the MontaVista Software, Inc. nor the names of its * contributors may be used to endorse or promote products derived from this * software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF * THE POSSIBILITY OF SUCH DAMAGE. */ /* * The first version of this code was based upon Yair Amir's PhD thesis: * http://www.cs.jhu.edu/~yairamir/phd.ps) (ch4,5). * * The current version of totemsrp implements the Totem protocol specified in: * http://citeseer.ist.psu.edu/amir95totem.html * * The deviations from the above published protocols are: * - encryption of message contents with SOBER128 * - authentication of meessage contents with SHA1/HMAC * - token hold mode where token doesn't rotate on unused ring - reduces cpu * usage on 1.6ghz xeon from 35% to less then .1 % as measured by top */ #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "aispoll.h" #include "totemsrp.h" #include "../include/queue.h" #include "../include/sq.h" #include "../include/list.h" #include "hdb.h" #include "swab.h" #include "crypto.h" #define AUTHENTICATION 1 /* use authentication */ #define ENCRYPTION 1 /* use encryption */ #define LOCALHOST_IP inet_addr("127.0.0.1") #define QUEUE_RTR_ITEMS_SIZE_MAX 2000 /* allow 512 retransmit items */ #define NEW_MESSAGE_QUEUE_SIZE_MAX 2000 /* allow 500 messages to be queued */ #define RETRANS_MESSAGE_QUEUE_SIZE_MAX 2000 /* allow 500 messages to be queued */ #define RECEIVED_MESSAGE_QUEUE_SIZE_MAX 2000 /* allow 500 messages to be queued */ #define MAXIOVS 5 #define RETRANSMIT_ENTRIES_MAX 30 #define MISSING_MCAST_WINDOW 128 #define TIMEOUT_STATE_GATHER_JOIN 100 #define TIMEOUT_STATE_GATHER_CONSENSUS 200 #define TOKEN_RETRANSMITS_BEFORE_LOSS 4 #define TIMEOUT_TOKEN 200 #define TIMEOUT_TOKEN_RETRANSMIT (int)(TIMEOUT_TOKEN / (TOKEN_RETRANSMITS_BEFORE_LOSS + 0.2)) #define TIMEOUT_TOKEN_HOLD (int)(TIMEOUT_TOKEN_RETRANSMIT * 0.8 - (1000/HZ)) #define TIMEOUT_MERGE_DETECT 200 #define PACKET_SIZE_MAX 2000 #define FAIL_TO_RECV_CONST 250 #define SEQNO_UNCHANGED_CONST 20 #define TIMEOUT_DOWNCHECK 1000 void memb_set_print (char *string, struct in_addr *list, int list_entries); /* * we compare incoming messages to determine if their endian is * different - if so convert them * * do not change */ #define ENDIAN_LOCAL 0xff22 /* * Authentication of messages */ hmac_state totemsrp_hmac_state; prng_state totemsrp_prng_state; unsigned char totemsrp_private_key[1024]; unsigned int totemsrp_private_key_len; int stats_sent = 0; int stats_recv = 0; int stats_delv = 0; int stats_remcasts = 0; int stats_orf_token = 0; struct timeval stats_tv_start = { 0, 0 }; /* * Flow control mcasts and remcasts on last and current orf_token */ int fcc_remcast_last = 0; int fcc_mcast_last = 0; int fcc_mcast_current = 0; int fcc_remcast_current = 0; enum message_type { MESSAGE_TYPE_ORF_TOKEN = 0, /* Ordering, Reliability, Flow (ORF) control Token */ MESSAGE_TYPE_MCAST = 1, /* ring ordered multicast message */ MESSAGE_TYPE_MEMB_MERGE_DETECT = 2, /* merge rings if there are available rings */ MESSAGE_TYPE_MEMB_JOIN = 3, /* membership join message */ MESSAGE_TYPE_MEMB_COMMIT_TOKEN = 4, /* membership commit token */ MESSAGE_TYPE_TOKEN_HOLD_CANCEL = 5, /* cancel the holding of the token */ }; /* * New membership algorithm local variables */ struct consensus_list_item { struct in_addr addr; int set; }; static struct consensus_list_item consensus_list[PROCESSOR_COUNT_MAX]; static int consensus_list_entries; static struct in_addr my_proc_list[PROCESSOR_COUNT_MAX]; static struct in_addr my_failed_list[PROCESSOR_COUNT_MAX]; static struct in_addr my_new_memb_list[PROCESSOR_COUNT_MAX]; static struct in_addr my_trans_memb_list[PROCESSOR_COUNT_MAX]; static struct in_addr my_memb_list[PROCESSOR_COUNT_MAX]; static struct in_addr my_deliver_memb_list[PROCESSOR_COUNT_MAX]; static int my_proc_list_entries = 0; static int my_failed_list_entries = 0; static int my_new_memb_entries = 0; static int my_trans_memb_entries = 0; static int my_memb_entries = 0; static int my_deliver_memb_entries = 0; static struct memb_ring_id my_ring_id; static struct memb_ring_id my_old_ring_id; static int my_aru_count = 0; static int my_merge_detect_timeout_outstanding = 0; static int my_last_aru = 0; static int my_seq_unchanged = 0; static int my_received_flg = 1; static int my_high_seq_received = 0; static int my_install_seq = 0; static int my_rotation_counter = 0; static int my_set_retrans_flg = 0; static int my_retrans_flg_count = 0; static unsigned int my_high_ring_delivered = 0; static unsigned int timeout_token = TIMEOUT_TOKEN; static unsigned int timeout_token_retransmit = TIMEOUT_TOKEN_RETRANSMIT; static unsigned int timeout_token_hold = 0; static unsigned int token_retransmits_before_loss = TOKEN_RETRANSMITS_BEFORE_LOSS; static unsigned int timeout_state_gather_join = TIMEOUT_STATE_GATHER_JOIN; static unsigned int timeout_state_gather_consensus = TIMEOUT_STATE_GATHER_CONSENSUS; static unsigned int timeout_merge_detect = TIMEOUT_MERGE_DETECT; static unsigned int timeout_downcheck = TIMEOUT_DOWNCHECK; static unsigned int fail_to_recv_const = FAIL_TO_RECV_CONST; struct token_callback_instance { struct list_head list; int (*callback_fn) (enum totem_callback_token_type type, void *); enum totem_callback_token_type callback_type; int delete; void *data; }; /* * Queues used to order, deliver, and recover messages */ struct queue new_message_queue; struct queue retrans_message_queue; struct sq regular_sort_queue; struct sq recovery_sort_queue; /* * Multicast address */ struct sockaddr_in sockaddr_in_mcast; struct totemsrp_socket { int mcast; int token; }; /* * File descriptors in use by TOTEMSRP */ struct totemsrp_socket totemsrp_sockets[2]; /* * Received up to and including */ int my_aru = 0; static int my_high_delivered = 0; DECLARE_LIST_INIT (token_callback_received_listhead); DECLARE_LIST_INIT (token_callback_sent_listhead); char orf_token_retransmit[15000]; // sizeof (struct orf_token) + sizeof (struct rtr_item) * RETRANSMIT_ENTRIES_MAX]; int orf_token_retransmit_size; int my_token_seq = -1; /* * Timers */ poll_timer_handle timer_orf_token_timeout = 0; poll_timer_handle timer_orf_token_retransmit_timeout = 0; poll_timer_handle timer_orf_token_hold_retransmit_timeout = 0; poll_timer_handle timer_merge_detect_timeout = 0; poll_timer_handle memb_timer_state_gather_join_timeout = 0; poll_timer_handle memb_timer_state_gather_consensus_timeout = 0; poll_timer_handle memb_timer_state_commit_timeout = 0; poll_timer_handle timer_netif_check_timeout = 0; /* * Function called when new message received */ int (*totemsrp_recv) (char *group, struct iovec *iovec, int iov_len); /* * Function and data used to log messages */ static void (*totemsrp_log_printf) (int level, char *format, ...); int totemsrp_log_level_security; int totemsrp_log_level_error; int totemsrp_log_level_warning; int totemsrp_log_level_notice; int totemsrp_log_level_debug; #define HMAC_HASH_SIZE 20 struct security_header { unsigned char hash_digest[HMAC_HASH_SIZE]; /* The hash *MUST* be first in the data structure */ unsigned char salt[16]; /* random number */ } __attribute__((packed)); struct message_header { struct security_header security_header; char type; char encapsulated; // unsigned short filler; unsigned short endian_detector; } __attribute__((packed)); struct mcast { struct message_header header; int seq; int this_seqno; struct memb_ring_id ring_id; struct in_addr source; int guarantee; } __attribute__((packed)); /* * MTU - multicast message header - IP header - UDP header * * On lossy switches, making use of the DF UDP flag can lead to loss of * forward progress. So the packets must be fragmented by a higher layer * * This layer can only handle packets of MTU size. */ #define FRAGMENT_SIZE (PACKET_SIZE_MAX - sizeof (struct mcast) - 20 - 8) struct rtr_item { struct memb_ring_id ring_id; int seq; }__attribute__((packed)); struct orf_token { struct message_header header; int seq; int token_seq; int aru; struct in_addr aru_addr; struct memb_ring_id ring_id; short int fcc; int retrans_flg; int rtr_list_entries; struct rtr_item rtr_list[0]; }__attribute__((packed)); struct memb_join { struct message_header header; struct in_addr proc_list[PROCESSOR_COUNT_MAX]; int proc_list_entries; struct in_addr failed_list[PROCESSOR_COUNT_MAX]; int failed_list_entries; unsigned long long ring_seq; } __attribute__((packed)); struct memb_merge_detect { struct message_header header; struct memb_ring_id ring_id; } __attribute__((packed)); struct token_hold_cancel { struct message_header header; struct memb_ring_id ring_id; } __attribute__((packed)); struct memb_commit_token_memb_entry { struct memb_ring_id ring_id; int aru; int high_delivered; int received_flg; }__attribute__((packed)); struct memb_commit_token { struct message_header header; int token_seq; struct memb_ring_id ring_id; unsigned int retrans_flg; int memb_index; int addr_entries; struct in_addr addr[PROCESSOR_COUNT_MAX]; struct memb_commit_token_memb_entry memb_list[PROCESSOR_COUNT_MAX]; }__attribute__((packed)); struct message_item { struct mcast *mcast; struct iovec iovec[MAXIOVS]; int iov_len; }; struct sort_queue_item { struct iovec iovec[MAXIOVS]; int iov_len; }; enum memb_state { MEMB_STATE_OPERATIONAL = 1, MEMB_STATE_GATHER = 2, MEMB_STATE_COMMIT = 3, MEMB_STATE_RECOVERY = 4 }; static enum memb_state memb_state = MEMB_STATE_OPERATIONAL; static struct sockaddr_in my_id; struct sockaddr_in next_memb; static struct sockaddr_in memb_local_sockaddr_in; static char iov_buffer[15000]; //PACKET_SIZE_MAX]; static struct iovec totemsrp_iov_recv = { .iov_base = iov_buffer, .iov_len = sizeof (iov_buffer) }; static char iov_encrypted_buffer[15000]; //char orf_token_retransmit[15000]; // sizeof (struct orf_token) + sizeof (struct rtr_item) * RETRANSMIT_ENTRIES_MAX]; static struct iovec iov_encrypted = { .iov_base = iov_encrypted_buffer, .iov_len = sizeof (iov_encrypted_buffer) }; struct message_handlers { int count; int (*handler_functions[6]) (struct sockaddr_in *, struct iovec *, int, int, int); }; poll_handle *totemsrp_poll_handle; void (*totemsrp_deliver_fn) ( struct in_addr source_addr, struct iovec *iovec, int iov_len, int endian_conversion_required) = 0; void (*totemsrp_confchg_fn) ( enum totem_configuration_type configuration_type, struct in_addr *member_list, int member_list_entries, struct in_addr *left_list, int left_list_entries, struct in_addr *joined_list, int joined_list_entries, struct memb_ring_id *ring_id) = 0; static struct totem_interface *totemsrp_interfaces; static int totemsrp_interface_count; /* * forward decls */ static int message_handler_orf_token (struct sockaddr_in *, struct iovec *, int, int, int); static int message_handler_mcast (struct sockaddr_in *, struct iovec *, int, int, int); static int message_handler_memb_merge_detect (struct sockaddr_in *, struct iovec *, int, int, int); static int message_handler_memb_join (struct sockaddr_in *, struct iovec *, int, int, int); static int message_handler_memb_commit_token (struct sockaddr_in *, struct iovec *, int, int, int); static int message_handler_token_hold_cancel (struct sockaddr_in *, struct iovec *, int, int, int); static void memb_ring_id_create_or_load (struct memb_ring_id *); static int recv_handler (poll_handle handle, int fd, int revents, void *data, unsigned int *prio); static int netif_determine (struct sockaddr_in *bindnet, struct sockaddr_in *bound_to,int *interface_up); static int loopback_determine (struct sockaddr_in *bound_to); static void netif_down_check (void); static void token_callbacks_execute (enum totem_callback_token_type type); #define NETIF_STATE_REPORT_UP 1 #define NETIF_STATE_REPORT_DOWN 2 #define BIND_STATE_UNBOUND 0 #define BIND_STATE_REGULAR 1 #define BIND_STATE_LOOPBACK 2 int netif_state_report = NETIF_STATE_REPORT_UP | NETIF_STATE_REPORT_DOWN; int netif_bind_state = BIND_STATE_UNBOUND; static int totemsrp_build_sockets (struct sockaddr_in *sockaddr_mcast, struct sockaddr_in *sockaddr_bindnet, struct totemsrp_socket *sockets, struct sockaddr_in *bound_to, int *interface_up); static int totemsrp_build_sockets_loopback (struct sockaddr_in *sockaddr_mcast, struct sockaddr_in *sockaddr_bindnet, struct totemsrp_socket *sockets, struct sockaddr_in *bound_to); static void memb_state_gather_enter (void); static void messages_deliver_to_app (int skip, int end_point); static int orf_token_mcast (struct orf_token *oken, int fcc_mcasts_allowed, struct sockaddr_in *system_from); static int messages_free (int token_aru); static void encrypt_and_sign (struct iovec *iovec, int iov_len); static int authenticate_and_decrypt (struct iovec *iov); static int recv_handler (poll_handle handle, int fd, int revents, void *data, unsigned int *prio); static void memb_ring_id_store (struct memb_commit_token *commit_token); static void memb_state_commit_token_update (struct memb_commit_token *memb_commit_token); static int memb_state_commit_token_send (struct memb_commit_token *memb_commit_token); static void memb_state_commit_token_create (struct memb_commit_token *commit_token); static int token_hold_cancel_send (void); static void orf_token_endian_convert (struct orf_token *in, struct orf_token *out); static void memb_commit_token_endian_convert (struct memb_commit_token *in, struct memb_commit_token *out); static void memb_join_endian_convert (struct memb_join *in, struct memb_join *out); static void mcast_endian_convert (struct mcast *in, struct mcast *out); struct message_handlers totemsrp_message_handlers = { 6, { message_handler_orf_token, message_handler_mcast, message_handler_memb_merge_detect, message_handler_memb_join, message_handler_memb_commit_token, message_handler_token_hold_cancel } }; void totemsrp_log_printf_init ( void (*log_printf) (int , char *, ...), int log_level_security, int log_level_error, int log_level_warning, int log_level_notice, int log_level_debug) { totemsrp_log_level_security = log_level_security; totemsrp_log_level_error = log_level_error; totemsrp_log_level_warning = log_level_warning; totemsrp_log_level_notice = log_level_notice; totemsrp_log_level_debug = log_level_debug; totemsrp_log_printf = log_printf; } #ifdef CODE_COVERAGE_COMPILE_OUT void print_digest (char *where, unsigned char *digest) { int i; printf ("DIGEST %s:\n", where); for (i = 0; i < 16; i++) { printf ("%x ", digest[i]); } printf ("\n"); } void print_msg (unsigned char *msg, int size) { int i; printf ("MSG CONTENTS START\n"); for (i = 0; i < size; i++) { printf ("%x ", msg[i]); if ((i % 16) == 15) { printf ("\n"); } } printf ("MSG CONTENTS DONE\n"); } #endif /* * Exported interfaces */ int totemsrp_initialize ( struct openais_config *openais_config, poll_handle *poll_handle, unsigned char *private_key, int private_key_len, void *member_private, int member_private_len, void (*deliver_fn) ( struct in_addr source_addr, struct iovec *iovec, int iov_len, int endian_conversion_required), void (*confchg_fn) ( enum totem_configuration_type configuration_type, struct in_addr *member_list, int member_list_entries, struct in_addr *left_list, int left_list_entries, struct in_addr *joined_list, int joined_list_entries, struct memb_ring_id *ring_id)) { unsigned int *timeouts = openais_config->timeouts; int i; timeout_token_hold = (int)(timeout_token_retransmit * 0.8 - (1000/HZ)); /* * Initialize random number generator for later use to generate salt */ memcpy (totemsrp_private_key, private_key, private_key_len); totemsrp_private_key_len = private_key_len; rng_make_prng (128, PRNG_SOBER, &totemsrp_prng_state, NULL); /* * Initialize local variables for totemsrp */ memcpy (&sockaddr_in_mcast, &openais_config->mcast_addr, sizeof (struct sockaddr_in)); memset (&next_memb, 0, sizeof (struct sockaddr_in)); memset (iov_buffer, 0, PACKET_SIZE_MAX); /* * Update our timeout values if they were specified in the openais.conf * file. */ for (i = 0; i < MAX_TOTEM_TIMEOUTS; i++) { if (!timeouts[i]) { continue; } switch (i) { case TOTEM_TOKEN: timeout_token = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Overriding token timeout to (%u ms)\n", timeouts[i]); timeout_token_retransmit = (int)(timeout_token / (token_retransmits_before_loss + 0.2)); timeout_token_hold = (int)(timeout_token_retransmit * 0.8 - (1000/HZ)); break; case TOTEM_RETRANSMIT_TOKEN: timeout_token_retransmit = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Overriding token retransmit timeout to (%u ms)\n", timeouts[i]); break; case TOTEM_RETRANSMITS_BEFORE_LOSS: token_retransmits_before_loss = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Overriding retransmits before loss (%u retrans)\n", timeouts[i]); break; case TOTEM_HOLD_TOKEN: timeout_token_hold = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Overriding token hold timeout to (%u ms)\n", timeouts[i]); break; case TOTEM_JOIN: timeout_state_gather_join = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Join Timeout set to %u ms\n", timeouts[i]); break; case TOTEM_CONSENSUS: timeout_state_gather_consensus = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Consensus Timeout set to %u ms\n", timeouts[i]); break; case TOTEM_MERGE: timeout_merge_detect = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Merge Detect Timeout set to %u ms\n", timeouts[i]); break; case TOTEM_DOWNCHECK: timeout_downcheck = timeouts[i]; totemsrp_log_printf (totemsrp_log_level_notice, "Downcheck Timeout set to %u ms\n", timeouts[i]); break; case TOTEM_FAIL_RECV_CONST: totemsrp_log_printf (totemsrp_log_level_notice, "Failed To Receive Const set to %u\n", timeouts[i]); fail_to_recv_const = timeouts[i]; break; default: totemsrp_log_printf (totemsrp_log_level_notice, "Received unknown timeout type: %d\n", timeouts[i]); break; } } totemsrp_log_printf (totemsrp_log_level_notice, "Token Timeout (%d ms) retransmit timeout (%d ms)\n", timeout_token, timeout_token_retransmit); totemsrp_log_printf (totemsrp_log_level_notice, "token hold (%d ms) retransmits before loss (%d retrans)\n", timeout_token_hold, token_retransmits_before_loss); queue_init (&new_message_queue, NEW_MESSAGE_QUEUE_SIZE_MAX, sizeof (struct message_item)); queue_init (&retrans_message_queue, RETRANS_MESSAGE_QUEUE_SIZE_MAX, sizeof (struct message_item)); sq_init (®ular_sort_queue, QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0); sq_init (&recovery_sort_queue, QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0); totemsrp_interfaces = openais_config->interfaces; totemsrp_interface_count = 1; /* totemsrp_interface_count = openais_config->interface_count; */ totemsrp_poll_handle = poll_handle; netif_down_check(); memb_state_gather_enter (); totemsrp_deliver_fn = deliver_fn; totemsrp_confchg_fn = confchg_fn; return (0); } /* * Set operations for use by the membership algorithm */ static void memb_consensus_reset (void) { consensus_list_entries = 0; } void memb_set_subtract (struct in_addr *out_list, int *out_list_entries, struct in_addr *one_list, int one_list_entries, struct in_addr *two_list, int two_list_entries) { int found = 0; int i; int j; *out_list_entries = 0; for (i = 0; i < one_list_entries; i++) { for (j = 0; j < two_list_entries; j++) { if (one_list[i].s_addr == two_list[j].s_addr) { found = 1; break; } } if (found == 0) { out_list[*out_list_entries].s_addr = one_list[i].s_addr; *out_list_entries = *out_list_entries + 1; } found = 0; } } /* * Set consensus for a specific processor */ static void memb_consensus_set (struct in_addr *addr) { int found = 0; int i; for (i = 0; i < consensus_list_entries; i++) { if (addr->s_addr == consensus_list[i].addr.s_addr) { found = 1; break; /* found entry */ } } consensus_list[i].addr.s_addr = addr->s_addr; consensus_list[i].set = 1; if (found == 0) { consensus_list_entries++; } return; } /* * Is consensus set for a specific processor */ static int memb_consensus_isset (struct in_addr *addr) { int i; for (i = 0; i < consensus_list_entries; i++) { if (addr->s_addr == consensus_list[i].addr.s_addr) { return (consensus_list[i].set); } } return (0); } /* * Is consensus agreed upon based upon consensus database */ static int memb_consensus_agreed (void) { struct in_addr token_memb[PROCESSOR_COUNT_MAX]; int token_memb_entries = 0; int agreed = 1; int i; memb_set_subtract (token_memb, &token_memb_entries, my_proc_list, my_proc_list_entries, my_failed_list, my_failed_list_entries); for (i = 0; i < token_memb_entries; i++) { if (memb_consensus_isset (&token_memb[i]) == 0) { agreed = 0; break; } } assert (token_memb_entries >= 1); return (agreed); } void memb_consensus_notset (struct in_addr *no_consensus_list, int *no_consensus_list_entries, struct in_addr *comparison_list, int comparison_list_entries) { int i; *no_consensus_list_entries = 0; for (i = 0; i < my_proc_list_entries; i++) { if (memb_consensus_isset (&my_proc_list[i]) == 0) { no_consensus_list[*no_consensus_list_entries].s_addr = my_proc_list[i].s_addr; *no_consensus_list_entries = *no_consensus_list_entries + 1; } } } /* * Is set1 equal to set2 Entries can be in different orders */ int memb_set_equal (struct in_addr *set1, int set1_entries, struct in_addr *set2, int set2_entries) { int i; int j; int found = 0; if (set1_entries != set2_entries) { return (0); } for (i = 0; i < set2_entries; i++) { for (j = 0; j < set1_entries; j++) { if (set1[j].s_addr == set2[i].s_addr) { found = 1; break; } } if (found == 0) { return (0); } found = 0; } return (1); } /* * Is subset fully contained in fullset */ int memb_set_subset (struct in_addr *subset, int subset_entries, struct in_addr *fullset, int fullset_entries) { int i; int j; int found = 0; if (subset_entries > fullset_entries) { return (0); } for (i = 0; i < subset_entries; i++) { for (j = 0; j < fullset_entries; j++) { if (subset[i].s_addr == fullset[j].s_addr) { found = 1; } } if (found == 0) { return (0); } found = 1; } return (1); } /* * merge subset into fullset taking care not to add duplicates */ void memb_set_merge (struct in_addr *subset, int subset_entries, struct in_addr *fullset, int *fullset_entries) { int found = 0; int i; int j; for (i = 0; i < subset_entries; i++) { for (j = 0; j < *fullset_entries; j++) { if (fullset[j].s_addr == subset[i].s_addr) { found = 1; break; } } if (found == 0) { fullset[j].s_addr = subset[i].s_addr; *fullset_entries = *fullset_entries + 1; } found = 0; } return; } void memb_set_and (struct in_addr *set1, int set1_entries, struct in_addr *set2, int set2_entries, struct in_addr *and, int *and_entries) { int i; int j; int found = 0; *and_entries = 0; for (i = 0; i < set2_entries; i++) { for (j = 0; j < set1_entries; j++) { if (set1[j].s_addr == set2[i].s_addr) { found = 1; break; } } if (found) { and[*and_entries].s_addr = set1[j].s_addr; *and_entries = *and_entries + 1; } found = 0; } return; } void memb_set_print (char *string, struct in_addr *list, int list_entries) { int i; printf ("List '%s' contains %d entries:\n", string, list_entries); for (i = 0; i < list_entries; i++) { printf ("addr %s\n", inet_ntoa (list[i])); } } static void timer_function_orf_token_timeout (void *data); static void timer_function_token_retransmit_timeout (void *data); static void timer_function_token_hold_retransmit_timeout (void *data); static void timer_function_merge_detect_timeout (void *data); void reset_token_retransmit_timeout (void) { poll_timer_delete (*totemsrp_poll_handle, timer_orf_token_retransmit_timeout); poll_timer_add (*totemsrp_poll_handle, timeout_token_retransmit, 0, timer_function_token_retransmit_timeout, &timer_orf_token_retransmit_timeout); } void start_merge_detect_timeout (void) { if (my_merge_detect_timeout_outstanding == 0) { poll_timer_add (*totemsrp_poll_handle, timeout_merge_detect, 0, timer_function_merge_detect_timeout, &timer_merge_detect_timeout); my_merge_detect_timeout_outstanding = 1; } } void cancel_merge_detect_timeout (void) { poll_timer_delete (*totemsrp_poll_handle, timer_merge_detect_timeout); my_merge_detect_timeout_outstanding = 0; } /* * ring_state_* is used to save and restore the sort queue * state when a recovery operation fails (and enters gather) */ static int old_ring_state_saved = 0; static int old_ring_state_aru = 0; static int old_ring_state_high_seq_received = 0; static void old_ring_state_save (void) { if (old_ring_state_saved == 0) { old_ring_state_saved = 1; old_ring_state_aru = my_aru; old_ring_state_high_seq_received = my_high_seq_received; totemsrp_log_printf (totemsrp_log_level_notice, "Saving state aru %d high seq recieved %d\n", my_aru, my_high_seq_received); } } static int ring_saved = 0; static void ring_save (void) { if (ring_saved == 0) { ring_saved = 1; memcpy (&my_old_ring_id, &my_ring_id, sizeof (struct memb_ring_id)); } } static void ring_reset (void) { ring_saved = 0; } static void ring_state_restore (void) { if (old_ring_state_saved) { my_ring_id.rep.s_addr = 0; my_aru = old_ring_state_aru; my_high_seq_received = old_ring_state_high_seq_received; totemsrp_log_printf (totemsrp_log_level_debug, "Restoring my_aru %d my high seq received %d\n", my_aru, my_high_seq_received); } } static void old_ring_state_reset (void) { old_ring_state_saved = 0; } void reset_token_timeout (void) { poll_timer_delete (*totemsrp_poll_handle, timer_orf_token_timeout); poll_timer_add (*totemsrp_poll_handle, timeout_token, (void *)9999, timer_function_orf_token_timeout, &timer_orf_token_timeout); } void cancel_token_timeout (void) { poll_timer_delete (*totemsrp_poll_handle, timer_orf_token_timeout); } void cancel_token_retransmit_timeout (void) { poll_timer_delete (*totemsrp_poll_handle, timer_orf_token_retransmit_timeout); } void start_token_hold_retransmit_timeout (void) { poll_timer_add (*totemsrp_poll_handle, timeout_token_hold, (void *)9999, timer_function_token_hold_retransmit_timeout, &timer_orf_token_hold_retransmit_timeout); } void cancel_token_hold_retransmit_timeout (void) { poll_timer_delete (*totemsrp_poll_handle, timer_orf_token_hold_retransmit_timeout); } static void memb_state_consensus_timeout_expired (void) { struct in_addr no_consensus_list[PROCESSOR_COUNT_MAX]; int no_consensus_list_entries; if (memb_consensus_agreed ()) { memb_consensus_reset (); memb_consensus_set (&my_id.sin_addr); reset_token_timeout (); // REVIEWED } else { memb_consensus_notset (no_consensus_list, &no_consensus_list_entries, my_proc_list, my_proc_list_entries); memb_set_merge (no_consensus_list, no_consensus_list_entries, my_failed_list, &my_failed_list_entries); memb_state_gather_enter (); } } static int memb_join_message_send (void); static int memb_merge_detect_transmit (void); /* * Timers used for various states of the membership algorithm */ static void timer_function_orf_token_timeout (void *data) { totemsrp_log_printf (totemsrp_log_level_notice, "The token was lost in state %d from timer %x\n", memb_state, data); switch (memb_state) { case MEMB_STATE_OPERATIONAL: netif_down_check(); memb_state_gather_enter (); break; case MEMB_STATE_GATHER: memb_state_consensus_timeout_expired (); memb_state_gather_enter (); break; case MEMB_STATE_COMMIT: memb_state_gather_enter (); break; case MEMB_STATE_RECOVERY: ring_state_restore (); memb_state_gather_enter(); break; } } static void memb_timer_function_state_gather (void *data) { switch (memb_state) { case MEMB_STATE_OPERATIONAL: case MEMB_STATE_RECOVERY: assert (0); /* this should never happen */ break; case MEMB_STATE_GATHER: case MEMB_STATE_COMMIT: memb_join_message_send (); /* * Restart the join timeout `*/ poll_timer_delete (*totemsrp_poll_handle, memb_timer_state_gather_join_timeout); poll_timer_add (*totemsrp_poll_handle, timeout_state_gather_join, 0, memb_timer_function_state_gather, &memb_timer_state_gather_join_timeout); break; } } static void memb_timer_function_gather_consensus_timeout (void *data) { memb_state_consensus_timeout_expired (); } void deliver_messages_from_recovery_to_regular (void) { int i; struct sort_queue_item *recovery_message_item; struct sort_queue_item regular_message_item; int res; void *ptr; struct mcast *mcast; totemsrp_log_printf (totemsrp_log_level_debug, "recovery to regular %d-%d\n", 1, my_aru); /* * Move messages from recovery to regular sort queue */ // todo should i be initialized to 0 or 1 ? for (i = 1; i <= my_aru; i++) { res = sq_item_get (&recovery_sort_queue, i, &ptr); if (res != 0) { continue; } printf ("Transferring message with seq id %d\n", i); recovery_message_item = (struct sort_queue_item *)ptr; /* * Convert recovery message into regular message */ if (recovery_message_item->iov_len > 1) { mcast = recovery_message_item->iovec[1].iov_base; memcpy (®ular_message_item.iovec[0], &recovery_message_item->iovec[1], sizeof (struct iovec) * recovery_message_item->iov_len); } else { mcast = recovery_message_item->iovec[0].iov_base; totemsrp_log_printf (totemsrp_log_level_notice, "encapsulated is %d\n", mcast->header.encapsulated); if (mcast->header.encapsulated == 1) { /* * Message is a recovery message encapsulated * in a new ring message */ regular_message_item.iovec[0].iov_base = recovery_message_item->iovec[0].iov_base + sizeof (struct mcast); regular_message_item.iovec[0].iov_len = recovery_message_item->iovec[0].iov_len - sizeof (struct mcast); regular_message_item.iov_len = 1; mcast = regular_message_item.iovec[0].iov_base; } else { printf ("not encapsulated\n"); continue; /* TODO this case shouldn't happen */ /* * Message is originated on new ring and not * encapsulated */ regular_message_item.iovec[0].iov_base = recovery_message_item->iovec[0].iov_base; regular_message_item.iovec[0].iov_len = recovery_message_item->iovec[0].iov_len; } } totemsrp_log_printf (totemsrp_log_level_debug, "comparing if ring id is for this processors old ring seqno %d\n", mcast->seq); /* * Only add this message to the regular sort * queue if it was originated with the same ring * id as the previous ring */ if (memcmp (&my_old_ring_id, &mcast->ring_id, sizeof (struct memb_ring_id)) == 0) { totemsrp_log_printf (totemsrp_log_level_notice, "adding msg with seq no %d\n", mcast->seq, mcast->this_seqno); regular_message_item.iov_len = recovery_message_item->iov_len; res = sq_item_inuse (®ular_sort_queue, mcast->seq); if (res == 0) { sq_item_add (®ular_sort_queue, ®ular_message_item, mcast->seq); if (mcast->seq > old_ring_state_high_seq_received) { old_ring_state_high_seq_received = mcast->seq; } } } else { totemsrp_log_printf (totemsrp_log_level_notice, "-not adding msg with seq no %d\n", mcast->seq); } } } /* * Change states in the state machine of the membership algorithm */ static void memb_state_operational_enter (void) { struct in_addr joined_list[PROCESSOR_COUNT_MAX]; int joined_list_entries = 0; struct in_addr left_list[PROCESSOR_COUNT_MAX]; int left_list_entries = 0; int aru_save; old_ring_state_reset (); ring_reset (); deliver_messages_from_recovery_to_regular (); totemsrp_log_printf (totemsrp_log_level_debug, "Delivering to app %d to %d\n", my_high_delivered + 1, old_ring_state_high_seq_received); aru_save = my_aru; my_aru = old_ring_state_aru; messages_deliver_to_app (0, old_ring_state_high_seq_received); /* * Calculate joined and left list */ memb_set_subtract (left_list, &left_list_entries, my_memb_list, my_memb_entries, my_trans_memb_list, my_trans_memb_entries); memb_set_subtract (joined_list, &joined_list_entries, my_new_memb_list, my_new_memb_entries, my_trans_memb_list, my_trans_memb_entries); /* * Deliver transitional configuration to application */ totemsrp_confchg_fn (TOTEM_CONFIGURATION_TRANSITIONAL, my_trans_memb_list, my_trans_memb_entries, left_list, left_list_entries, 0, 0, &my_ring_id); // TODO we need to filter to ensure we only deliver those // messages which are part of my_deliver_memb messages_deliver_to_app (1, old_ring_state_high_seq_received); my_aru = aru_save; /* * Deliver regular configuration to application */ totemsrp_confchg_fn (TOTEM_CONFIGURATION_REGULAR, my_new_memb_list, my_new_memb_entries, 0, 0, joined_list, joined_list_entries, &my_ring_id); /* * Install new membership */ my_memb_entries = my_new_memb_entries; memcpy (my_memb_list, my_new_memb_list, sizeof (struct in_addr) * my_memb_entries); last_released = 0; my_set_retrans_flg = 0; /* * The recovery sort queue now becomes the regular * sort queue. It is necessary to copy the state * into the regular sort queue. */ sq_copy (®ular_sort_queue, &recovery_sort_queue); my_last_aru = 0; my_proc_list_entries = my_new_memb_entries; memcpy (my_proc_list, my_new_memb_list, sizeof (struct in_addr) * my_memb_entries); my_failed_list_entries = 0; my_high_delivered = my_aru; // TODO the recovery messages are leaked totemsrp_log_printf (totemsrp_log_level_notice, "entering OPERATIONAL state.\n"); memb_state = MEMB_STATE_OPERATIONAL; return; } static void memb_state_gather_enter (void) { memb_set_merge (&my_id.sin_addr, 1, my_proc_list, &my_proc_list_entries); memb_join_message_send (); /* * Restart the join timeout */ poll_timer_delete (*totemsrp_poll_handle, memb_timer_state_gather_join_timeout); poll_timer_add (*totemsrp_poll_handle, timeout_state_gather_join, 0, memb_timer_function_state_gather, &memb_timer_state_gather_join_timeout); /* * Restart the consensus timeout */ poll_timer_delete (*totemsrp_poll_handle, memb_timer_state_gather_consensus_timeout); poll_timer_add (*totemsrp_poll_handle, timeout_state_gather_consensus, 0, memb_timer_function_gather_consensus_timeout, &memb_timer_state_gather_consensus_timeout); /* * Cancel the token loss and token retransmission timeouts */ cancel_token_retransmit_timeout (); // REVIEWED cancel_token_timeout (); // REVIEWED cancel_merge_detect_timeout (); memb_consensus_reset (); memb_consensus_set (&my_id.sin_addr); totemsrp_log_printf (totemsrp_log_level_notice, "entering GATHER state.\n"); memb_state = MEMB_STATE_GATHER; return; } void timer_function_token_retransmit_timeout (void *data); static void memb_state_commit_enter (struct memb_commit_token *commit_token) { ring_save (); old_ring_state_save (); memb_state_commit_token_update (commit_token); memb_state_commit_token_send (commit_token); memb_ring_id_store (commit_token); poll_timer_delete (*totemsrp_poll_handle, memb_timer_state_gather_join_timeout); memb_timer_state_gather_join_timeout = 0; poll_timer_delete (*totemsrp_poll_handle, memb_timer_state_gather_consensus_timeout); memb_timer_state_gather_consensus_timeout = 0; reset_token_timeout (); // REVIEWED reset_token_retransmit_timeout (); // REVIEWED totemsrp_log_printf (totemsrp_log_level_notice, "entering COMMIT state.\n"); memb_state = MEMB_STATE_COMMIT; return; } void memb_state_recovery_enter (struct memb_commit_token *commit_token) { int i; #ifdef COMPILE_OUT int local_received_flg = 1; #endif unsigned int low_ring_aru; unsigned int messages_originated = 0; my_high_ring_delivered = 0; sq_reinit (&recovery_sort_queue, 0); queue_reinit (&retrans_message_queue); low_ring_aru = old_ring_state_high_seq_received; memb_state_commit_token_send (commit_token); my_token_seq = -1; /* * Build regular configuration */ my_new_memb_entries = commit_token->addr_entries; memcpy (my_new_memb_list, commit_token->addr, sizeof (struct in_addr) * my_new_memb_entries); /* * Build transitional configuration */ memb_set_and (my_new_memb_list, my_new_memb_entries, my_memb_list, my_memb_entries, my_trans_memb_list, &my_trans_memb_entries); for (i = 0; i < my_new_memb_entries; i++) { totemsrp_log_printf (totemsrp_log_level_notice, "position [%d] member %s:\n", i, inet_ntoa (commit_token->addr[i])); totemsrp_log_printf (totemsrp_log_level_notice, "previous ring seq %lld rep %s\n", commit_token->memb_list[i].ring_id.seq, inet_ntoa (commit_token->memb_list[i].ring_id.rep)); totemsrp_log_printf (totemsrp_log_level_notice, "aru %d high delivered %d received flag %d\n", commit_token->memb_list[i].aru, commit_token->memb_list[i].high_delivered, commit_token->memb_list[i].received_flg); assert (commit_token->memb_list[i].ring_id.rep.s_addr); } /* * Determine if any received flag is false */ #ifdef COMPILE_OUT for (i = 0; i < commit_token->addr_entries; i++) { if (memb_set_subset (&my_new_memb_list[i], 1, my_trans_memb_list, my_trans_memb_entries) && commit_token->memb_list[i].received_flg == 0) { #endif my_deliver_memb_entries = my_trans_memb_entries; memcpy (my_deliver_memb_list, my_trans_memb_list, sizeof (struct in_addr) * my_trans_memb_entries); #ifdef COMPILE_OUT local_received_flg = 0; break; } } #endif // if (local_received_flg == 0) { /* * Calculate my_low_ring_aru, my_high_ring_delivered for the transitional membership */ for (i = 0; i < commit_token->addr_entries; i++) { printf ("comparing %d old ring %s.%lld with commit ring %s.%lld.\n", i, inet_ntoa (my_old_ring_id.rep), my_old_ring_id.seq, inet_ntoa (commit_token->memb_list[i].ring_id.rep), commit_token->memb_list[i].ring_id.seq); printf ("memb set subset %d\n", memb_set_subset (&my_new_memb_list[i], 1, my_deliver_memb_list, my_deliver_memb_entries)); if (memb_set_subset (&my_new_memb_list[i], 1, my_deliver_memb_list, my_deliver_memb_entries) && memcmp (&my_old_ring_id, &commit_token->memb_list[i].ring_id, sizeof (struct memb_ring_id)) == 0) { if (low_ring_aru == 0 || low_ring_aru > commit_token->memb_list[i].aru) { low_ring_aru = commit_token->memb_list[i].aru; } if (my_high_ring_delivered < commit_token->memb_list[i].high_delivered) { my_high_ring_delivered = commit_token->memb_list[i].high_delivered; } } } assert (low_ring_aru != 0xffffffff); /* * Cpy all old ring messages to retrans_message_queue */ totemsrp_log_printf (totemsrp_log_level_notice, "copying all old ring messages from %d-%d.\n", low_ring_aru + 1, old_ring_state_high_seq_received); for (i = low_ring_aru + 1; i <= old_ring_state_high_seq_received; i++) { struct sort_queue_item *sort_queue_item; struct message_item message_item; void *ptr; int res; res = sq_item_get (®ular_sort_queue, i, &ptr); if (res != 0) { printf ("-not copying %d-\n", i); continue; } printf ("copying %d\n", i); sort_queue_item = ptr; assert (sort_queue_item->iov_len > 0); assert (sort_queue_item->iov_len <= MAXIOVS); messages_originated++; memset (&message_item, 0, sizeof (struct message_item)); // TODO LEAK message_item.mcast = malloc (sizeof (struct mcast)); assert (message_item.mcast); memcpy (message_item.mcast, sort_queue_item->iovec[0].iov_base, sizeof (struct mcast)); memcpy (&message_item.mcast->ring_id, &my_ring_id, sizeof (struct memb_ring_id)); message_item.mcast->header.encapsulated = 1; message_item.iov_len = sort_queue_item->iov_len; memcpy (&message_item.iovec, &sort_queue_item->iovec, sizeof (struct iovec) * sort_queue_item->iov_len); queue_item_add (&retrans_message_queue, &message_item); } totemsrp_log_printf (totemsrp_log_level_notice, "Originated %d messages in RECOVERY.\n", messages_originated); // } my_aru = 0; my_aru_count = 0; my_seq_unchanged = 0; my_high_seq_received = 0; my_install_seq = 0; totemsrp_log_printf (totemsrp_log_level_notice, "entering RECOVERY state.\n"); reset_token_timeout (); // REVIEWED reset_token_retransmit_timeout (); // REVIEWED memb_state = MEMB_STATE_RECOVERY; return; } static void encrypt_and_sign (struct iovec *iovec, int iov_len) { char *addr = iov_encrypted.iov_base + sizeof (struct security_header); int i; char keys[48]; struct security_header *header = iov_encrypted.iov_base; prng_state keygen_prng_state; prng_state stream_prng_state; char *hmac_key = &keys[32]; char *cipher_key = &keys[16]; char *initial_vector = &keys[0]; unsigned long len; iov_encrypted.iov_len = 0; memset (keys, 0, sizeof (keys)); memset (header->salt, 0, sizeof (header->salt)); #if (defined(ENCRYPTION) || defined(AUTHENITCATION)) /* * Generate MAC, CIPHER, IV keys from private key */ sober128_read (header->salt, sizeof (header->salt), &totemsrp_prng_state); sober128_start (&keygen_prng_state); sober128_add_entropy (totemsrp_private_key, totemsrp_private_key_len, &keygen_prng_state); sober128_add_entropy (header->salt, sizeof (header->salt), &keygen_prng_state); sober128_read (keys, sizeof (keys), &keygen_prng_state); #endif #ifdef ENCRYPTION /* * Setup stream cipher */ sober128_start (&stream_prng_state); sober128_add_entropy (cipher_key, 16, &stream_prng_state); sober128_add_entropy (initial_vector, 16, &stream_prng_state); #endif #ifdef CODE_COVERAGE_COMPILE_OUT if (log_digest) { printf ("new encryption\n"); print_digest ("salt", header->salt); print_digest ("initial_vector", initial_vector); print_digest ("cipher_key", cipher_key); print_digest ("hmac_key", hmac_key); } #endif /* * Copy header of message, then remainder of message, then encrypt it */ memcpy (addr, iovec[0].iov_base + sizeof (struct security_header), iovec[0].iov_len - sizeof (struct security_header)); addr += iovec[0].iov_len - sizeof (struct security_header); iov_encrypted.iov_len += iovec[0].iov_len; for (i = 1; i < iov_len; i++) { memcpy (addr, iovec[i].iov_base, iovec[i].iov_len); addr += iovec[i].iov_len; iov_encrypted.iov_len += iovec[i].iov_len; } /* * Encrypt message by XORing stream cipher data */ #ifdef ENCRYPTION sober128_read (iov_encrypted.iov_base + sizeof (struct security_header), iov_encrypted.iov_len - sizeof (struct security_header), &stream_prng_state); #endif #ifdef AUTHENTICATION memset (&totemsrp_hmac_state, 0, sizeof (hmac_state)); /* * Sign the contents of the message with the hmac key and store signature in message */ hmac_init (&totemsrp_hmac_state, DIGEST_SHA1, hmac_key, 16); hmac_process (&totemsrp_hmac_state, iov_encrypted.iov_base + HMAC_HASH_SIZE, iov_encrypted.iov_len - HMAC_HASH_SIZE); len = hash_descriptor[DIGEST_SHA1]->hashsize; hmac_done (&totemsrp_hmac_state, header->hash_digest, &len); #endif #ifdef COMPILE_OUT print_digest ("initial_vector", initial_vector); print_digest ("cipher_key", cipher_key); print_digest ("hmac_key", hmac_key); print_digest ("salt", header->salt); print_digest ("sent digest", header->hash_digest); #endif } /* * Only designed to work with a message with one iov */ static int authenticate_and_decrypt (struct iovec *iov) { char keys[48]; struct security_header *header = iov[0].iov_base; prng_state keygen_prng_state; prng_state stream_prng_state; char *hmac_key = &keys[32]; char *cipher_key = &keys[16]; char *initial_vector = &keys[0]; char digest_comparison[HMAC_HASH_SIZE]; unsigned long len; int res = 0; iov_encrypted.iov_len = 0; #ifdef COMPILE_OUT printf ("Decryption message\n"); print_msg (header, iov[0].iov_len); #endif #if (defined(ENCRYPTION) || defined(AUTHENITCATION)) /* * Generate MAC, CIPHER, IV keys from private key */ memset (keys, 0, sizeof (keys)); sober128_start (&keygen_prng_state); sober128_add_entropy (totemsrp_private_key, totemsrp_private_key_len, &keygen_prng_state); sober128_add_entropy (header->salt, sizeof (header->salt), &keygen_prng_state); sober128_read (keys, sizeof (keys), &keygen_prng_state); #endif #ifdef ENCRYPTION /* * Setup stream cipher */ sober128_start (&stream_prng_state); sober128_add_entropy (cipher_key, 16, &stream_prng_state); sober128_add_entropy (initial_vector, 16, &stream_prng_state); #endif #ifdef CODE_COVERAGE_COMPILE_OUT if (log_digest) { printf ("New decryption\n"); print_digest ("salt", header->salt); print_digest ("initial_vector", initial_vector); print_digest ("cipher_key", cipher_key); print_digest ("hmac_key", hmac_key); } #endif #ifdef AUTHENTICATION /* * Authenticate contents of message */ hmac_init (&totemsrp_hmac_state, DIGEST_SHA1, hmac_key, 16); hmac_process (&totemsrp_hmac_state, iov->iov_base + HMAC_HASH_SIZE, iov->iov_len - HMAC_HASH_SIZE); len = hash_descriptor[DIGEST_SHA1]->hashsize; assert (HMAC_HASH_SIZE >= len); hmac_done (&totemsrp_hmac_state, digest_comparison, &len); #ifdef PRINTDIGESTS print_digest ("received digest", header->hash_digest); print_digest ("calculated digest", digest_comparison); #endif if (memcmp (digest_comparison, header->hash_digest, len) != 0) { #ifdef CODE_COVERAGE_COMPILE_OUT print_digest ("initial_vector", initial_vector); print_digest ("cipher_key", cipher_key); print_digest ("hmac_key", hmac_key); print_digest ("salt", header->salt); print_digest ("sent digest", header->hash_digest); print_digest ("calculated digest", digest_comparison); printf ("received message size %d\n", iov->iov_len); #endif totemsrp_log_printf (totemsrp_log_level_security, "Received message has invalid digest... ignoring.\n"); res = -1; return (-1); } #endif /* AUTHENTICATION */ /* * Decrypt the contents of the message with the cipher key */ #ifdef ENCRYPTION sober128_read (iov->iov_base + sizeof (struct security_header), iov->iov_len - sizeof (struct security_header), &stream_prng_state); #endif return (res); return (0); } void totemsrp_new_msg_signal (void) { token_hold_cancel_send (); } int totemsrp_mcast ( struct iovec *iovec, int iov_len, int guarantee) { int i; int j; struct message_item message_item; if (queue_is_full (&new_message_queue)) { return (-1); } for (j = 0, i = 0; i < iov_len; i++) { j+= iovec[i].iov_len; } memset (&message_item, 0, sizeof (struct message_item)); /* * Allocate pending item */ // TODO LEAK message_item.mcast = malloc (sizeof (struct mcast)); if (message_item.mcast == 0) { goto error_mcast; } /* * Set mcast header */ message_item.mcast->header.type = MESSAGE_TYPE_MCAST; message_item.mcast->header.endian_detector = ENDIAN_LOCAL; message_item.mcast->header.encapsulated = 2; message_item.mcast->guarantee = guarantee; message_item.mcast->source.s_addr = my_id.sin_addr.s_addr; for (i = 0; i < iov_len; i++) { // TODO LEAK message_item.iovec[i].iov_base = malloc (iovec[i].iov_len); if (message_item.iovec[i].iov_base == 0) { goto error_iovec; } memcpy (message_item.iovec[i].iov_base, iovec[i].iov_base, iovec[i].iov_len); message_item.iovec[i].iov_len = iovec[i].iov_len; } message_item.iov_len = iov_len; totemsrp_log_printf (totemsrp_log_level_debug, "mcasted message added to pending queue\n"); queue_item_add (&new_message_queue, &message_item); return (0); error_iovec: for (j = 0; j < i; j++) { free (message_item.iovec[j].iov_base); } return (-1); error_mcast: return (0); } /* * Determine if there is room to queue a new message */ int totemsrp_avail (void) { int avail; queue_avail (&new_message_queue, &avail); return (avail); } static int netif_determine (struct sockaddr_in *bindnet, struct sockaddr_in *bound_to, int *interface_up) { struct sockaddr_in *sockaddr_in; int id_fd; struct ifconf ifc; int numreqs = 0; int res; int i; in_addr_t mask_addr; *interface_up = 0; /* * Generate list of local interfaces in ifc.ifc_req structure */ id_fd = socket (AF_INET, SOCK_STREAM, 0); ifc.ifc_buf = 0; do { numreqs += 32; ifc.ifc_len = sizeof (struct ifreq) * numreqs; ifc.ifc_buf = (void *)realloc(ifc.ifc_buf, ifc.ifc_len); res = ioctl (id_fd, SIOCGIFCONF, &ifc); if (res < 0) { close (id_fd); return -1; } } while (ifc.ifc_len == sizeof (struct ifreq) * numreqs); res = -1; /* * Find interface address to bind to */ for (i = 0; i < ifc.ifc_len / sizeof (struct ifreq); i++) { sockaddr_in = (struct sockaddr_in *)&ifc.ifc_ifcu.ifcu_req[i].ifr_ifru.ifru_addr; mask_addr = inet_addr ("255.255.255.0"); if ((sockaddr_in->sin_family == AF_INET) && (sockaddr_in->sin_addr.s_addr & mask_addr) == (bindnet->sin_addr.s_addr & mask_addr)) { bound_to->sin_addr.s_addr = sockaddr_in->sin_addr.s_addr; res = i; if (ioctl(id_fd, SIOCGIFFLAGS, &ifc.ifc_ifcu.ifcu_req[i]) < 0) { printf ("couldn't do ioctl\n"); } *interface_up = ifc.ifc_ifcu.ifcu_req[i].ifr_ifru.ifru_flags & IFF_UP; break; /* for */ } } free (ifc.ifc_buf); close (id_fd); return (res); } static int loopback_determine (struct sockaddr_in *bound_to) { bound_to->sin_addr.s_addr = LOCALHOST_IP; if (&bound_to->sin_addr.s_addr == 0) { return -1; } return 1; } int firstrun = 0; /* * If the interface is up, the sockets for gmi are built. If the interface is down * this function is requeued in the timer list to retry building the sockets later. */ static void timer_function_netif_check_timeout () { int res; int interface_no; int interface_up; /* * Build sockets for every interface */ for (interface_no = 0; interface_no < totemsrp_interface_count; interface_no++) { netif_determine(&totemsrp_interfaces[interface_no].bindnet, &totemsrp_interfaces[interface_no].boundto, &interface_up); if (((netif_bind_state & BIND_STATE_LOOPBACK) && (!interface_up)) || ((netif_bind_state & BIND_STATE_REGULAR) && (interface_up))) { break; } totemsrp_log_printf(totemsrp_log_level_debug,"network interface UP %s\n", inet_ntoa (totemsrp_interfaces[interface_no].boundto.sin_addr)); if (totemsrp_sockets[interface_no].mcast > 0) { close (totemsrp_sockets[interface_no].mcast); poll_dispatch_delete (*totemsrp_poll_handle, totemsrp_sockets[interface_no].mcast); } if (totemsrp_sockets[interface_no].token > 0) { close (totemsrp_sockets[interface_no].token); poll_dispatch_delete (*totemsrp_poll_handle, totemsrp_sockets[interface_no].token); } if (!interface_up) { totemsrp_log_printf (totemsrp_log_level_notice,"Interface is down binding to LOOPBACK addr.\n"); netif_bind_state = BIND_STATE_LOOPBACK; res = totemsrp_build_sockets_loopback(&sockaddr_in_mcast, &totemsrp_interfaces[interface_no].bindnet, &totemsrp_sockets[interface_no], &totemsrp_interfaces[interface_no].boundto); totemsrp_log_printf (totemsrp_log_level_notice,"network interface LOCAL %s\n", inet_ntoa (totemsrp_interfaces[interface_no].boundto.sin_addr)); poll_dispatch_add (*totemsrp_poll_handle, totemsrp_sockets[interface_no].token, POLLIN, 0, recv_handler, UINT_MAX); continue; } netif_bind_state = BIND_STATE_REGULAR; memcpy(&sockaddr_in_mcast,&config_mcast_addr, sizeof (struct sockaddr_in)); /* * Create and bind the multicast and unicast sockets */ res = totemsrp_build_sockets (&sockaddr_in_mcast, &totemsrp_interfaces[interface_no].bindnet, &totemsrp_sockets[interface_no], &totemsrp_interfaces[interface_no].boundto, &interface_up); poll_dispatch_add (*totemsrp_poll_handle, totemsrp_sockets[interface_no].mcast, POLLIN, 0, recv_handler, UINT_MAX); poll_dispatch_add (*totemsrp_poll_handle, totemsrp_sockets[interface_no].token, POLLIN, 0, recv_handler, UINT_MAX); } memcpy (&my_id, &totemsrp_interfaces->boundto, sizeof (struct sockaddr_in)); /* * This stuff depends on totemsrp_build_sockets */ if (firstrun == 0) { firstrun += 1; memcpy (&my_memb_list[0], &totemsrp_interfaces->boundto, sizeof (struct sockaddr_in)); memb_ring_id_create_or_load (&my_ring_id); totemsrp_log_printf (totemsrp_log_level_notice, "Created or loaded sequence id %lld.%s for this ring.\n", my_ring_id.seq, inet_ntoa (my_ring_id.rep)); } if (interface_up) { if (netif_state_report & NETIF_STATE_REPORT_UP) { totemsrp_log_printf (totemsrp_log_level_notice, " The network interface is now up.\n"); netif_state_report = NETIF_STATE_REPORT_DOWN; memb_state_gather_enter (); } /* * If this is a single processor, detect downs which may not * be detected by token loss when the interface is downed */ if (my_memb_entries <= 1) { poll_timer_add (*totemsrp_poll_handle, timeout_downcheck, (void *)1, timer_function_netif_check_timeout, &timer_netif_check_timeout); } } else { if (netif_state_report & NETIF_STATE_REPORT_DOWN) { totemsrp_log_printf (totemsrp_log_level_notice, "The network interface is down.\n"); memb_state_gather_enter (); } netif_state_report = NETIF_STATE_REPORT_UP; /* * Add a timer to retry building interfaces and request memb_gather_enter */ cancel_token_retransmit_timeout (); cancel_token_timeout (); poll_timer_add (*totemsrp_poll_handle, timeout_downcheck, (void *)1, timer_function_netif_check_timeout, &timer_netif_check_timeout); } } /* * Check if an interface is down and reconfigure * totemsrp waiting for it to come back up */ static void netif_down_check (void) { timer_function_netif_check_timeout (); } static int totemsrp_build_sockets_loopback (struct sockaddr_in *sockaddr_mcast, struct sockaddr_in *sockaddr_bindnet, struct totemsrp_socket *sockets, struct sockaddr_in *bound_to) { struct ip_mreq mreq; struct sockaddr_in sockaddr_in; int res; memset (&mreq, 0, sizeof (struct ip_mreq)); /* * Determine the ip address bound to and the interface name */ res = loopback_determine (bound_to); if (res == -1) { return (-1); } /* TODO this should be somewhere else */ memb_local_sockaddr_in.sin_addr.s_addr = bound_to->sin_addr.s_addr; memb_local_sockaddr_in.sin_family = AF_INET; memb_local_sockaddr_in.sin_port = sockaddr_mcast->sin_port; sockaddr_in.sin_family = AF_INET; sockaddr_in.sin_port = sockaddr_mcast->sin_port; /* * Setup unicast socket */ sockets->token = socket (AF_INET, SOCK_DGRAM, 0); if (sockets->token == -1) { perror ("socket2"); return (-1); } /* * Bind to unicast socket used for token send/receives * This has the side effect of binding to the correct interface */ sockaddr_in.sin_addr.s_addr = bound_to->sin_addr.s_addr; res = bind (sockets->token, (struct sockaddr *)&sockaddr_in, sizeof (struct sockaddr_in)); if (res == -1) { perror ("bind2 failed"); return (-1); } memcpy(&sockaddr_in_mcast, &sockaddr_in, sizeof(struct sockaddr_in)); sockets->mcast = sockets->token; return (0); } static int totemsrp_build_sockets (struct sockaddr_in *sockaddr_mcast, struct sockaddr_in *sockaddr_bindnet, struct totemsrp_socket *sockets, struct sockaddr_in *bound_to, int *interface_up) { struct ip_mreq mreq; struct sockaddr_in sockaddr_in; char flag; int res; memset (&mreq, 0, sizeof (struct ip_mreq)); /* * Determine the ip address bound to and the interface name */ res = netif_determine (sockaddr_bindnet, bound_to, interface_up); if (res == -1) { return (-1); } /* TODO this should be somewhere else */ memb_local_sockaddr_in.sin_addr.s_addr = bound_to->sin_addr.s_addr; memb_local_sockaddr_in.sin_family = AF_INET; memb_local_sockaddr_in.sin_port = sockaddr_mcast->sin_port; /* * Create multicast socket */ sockets->mcast = socket (AF_INET, SOCK_DGRAM, 0); if (sockets->mcast == -1) { perror ("socket"); return (-1); } if (setsockopt (sockets->mcast, SOL_IP, IP_MULTICAST_IF, &bound_to->sin_addr, sizeof (struct in_addr)) < 0) { totemsrp_log_printf (totemsrp_log_level_warning, "Could not bind to device for multicast, group messaging may not work properly. (%s)\n", strerror (errno)); } /* * Bind to multicast socket used for multicast send/receives */ sockaddr_in.sin_family = AF_INET; sockaddr_in.sin_addr.s_addr = sockaddr_mcast->sin_addr.s_addr; sockaddr_in.sin_port = sockaddr_mcast->sin_port; res = bind (sockets->mcast, (struct sockaddr *)&sockaddr_in, sizeof (struct sockaddr_in)); if (res == -1) { perror ("bind failed"); return (-1); } /* * Setup unicast socket */ sockets->token = socket (AF_INET, SOCK_DGRAM, 0); if (sockets->token == -1) { perror ("socket2"); return (-1); } /* * Bind to unicast socket used for token send/receives * This has the side effect of binding to the correct interface */ sockaddr_in.sin_addr.s_addr = bound_to->sin_addr.s_addr; res = bind (sockets->token, (struct sockaddr *)&sockaddr_in, sizeof (struct sockaddr_in)); if (res == -1) { perror ("bind2 failed"); return (-1); } #ifdef CONFIG_USE_BROADCAST /* This config option doesn't work */ { int on = 1; setsockopt (sockets->mcast, SOL_SOCKET, SO_BROADCAST, (char *)&on, sizeof (on)); } #else /* * Join group membership on socket */ mreq.imr_multiaddr.s_addr = sockaddr_mcast->sin_addr.s_addr; mreq.imr_interface.s_addr = bound_to->sin_addr.s_addr; res = setsockopt (sockets->mcast, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof (mreq)); if (res == -1) { perror ("join multicast group failed"); return (-1); } #endif /* * Turn on multicast loopback */ flag = 1; res = setsockopt (sockets->mcast, IPPROTO_IP, IP_MULTICAST_LOOP, &flag, sizeof (flag)); if (res == -1) { perror ("turn off loopback"); return (-1); } return (0); } /* * Misc Management */ int in_addr_compare (const void *a, const void *b) { struct in_addr *in_addr_a = (struct in_addr *)a; struct in_addr *in_addr_b = (struct in_addr *)b; return (in_addr_a->s_addr > in_addr_b->s_addr); } /* * ORF Token Management */ /* * Recast message to mcast group if it is available */ int orf_token_remcast (int seq) { struct msghdr msg_mcast; struct sort_queue_item *sort_queue_item; int res; struct mcast *mcast; void *ptr; struct sq *sort_queue; if (memb_state == MEMB_STATE_RECOVERY) { sort_queue = &recovery_sort_queue; } else { sort_queue = ®ular_sort_queue; } /* * Get RTR item at seq, if not available, return */ res = sq_item_get (sort_queue, seq, &ptr); if (res != 0) { return -1; } sort_queue_item = ptr; mcast = (struct mcast *)sort_queue_item->iovec[0].iov_base; encrypt_and_sign (sort_queue_item->iovec, sort_queue_item->iov_len); /* * Build multicast message */ msg_mcast.msg_name = (caddr_t)&sockaddr_in_mcast; msg_mcast.msg_namelen = sizeof (struct sockaddr_in); msg_mcast.msg_iov = &iov_encrypted; msg_mcast.msg_iovlen = 1; msg_mcast.msg_control = 0; msg_mcast.msg_controllen = 0; msg_mcast.msg_flags = 0; /* * Multicast message */ res = sendmsg (totemsrp_sockets[0].mcast, &msg_mcast, MSG_NOSIGNAL | MSG_DONTWAIT); if (res == -1) { return (-1); } stats_sent += res; return (0); } /* * Free all freeable messages from ring */ static int messages_free (int token_aru) { struct sort_queue_item *regular_message; int i, j; int res; int log_release = 0; int release_to; release_to = token_aru; if (release_to > my_last_aru) { release_to = my_last_aru; } if (release_to > my_high_delivered) { release_to = my_high_delivered; } /* * Release retransmit list items if group aru indicates they are transmitted */ for (i = last_released; i <= release_to; i++) { void *ptr; res = sq_item_get (®ular_sort_queue, i, &ptr); if (res == 0) { regular_message = ptr; for (j = 0; j < regular_message->iov_len; j++) { free (regular_message->iovec[j].iov_base); } } sq_items_release (®ular_sort_queue, i); last_released = i + 1; log_release = 1; } if (log_release) { totemsrp_log_printf (totemsrp_log_level_debug, "releasing messages up to and including %d\n", release_to); } return (0); } void update_aru (void) { int i; int res; struct sq *sort_queue; if (memb_state == MEMB_STATE_RECOVERY) { sort_queue = &recovery_sort_queue; } else { sort_queue = ®ular_sort_queue; } for (i = my_aru + 1; i <= my_high_seq_received; i++) { void *ptr; res = sq_item_get (sort_queue, i, &ptr); /* * If hole, stop assembly */ if (res != 0) { break; } my_aru = i; } // totemsrp_log_printf (totemsrp_log_level_debug, // "setting received flag to FALSE %d %d\n", // my_aru, my_high_seq_received); my_received_flg = 0; if (my_aru == my_high_seq_received) { // totemsrp_log_printf (totemsrp_log_level_debug, // "setting received flag to TRUE %d %d\n", // my_aru, my_high_seq_received); my_received_flg = 1; } } /* * Multicasts pending messages onto the ring (requires orf_token possession) */ static int orf_token_mcast ( struct orf_token *token, int fcc_mcasts_allowed, struct sockaddr_in *system_from) { struct msghdr msg_mcast; struct sort_queue_item sort_queue_item; struct message_item *message_item = 0; int res = 0; struct mcast *mcast; struct queue *mcast_queue; struct sq *sort_queue; if (memb_state == MEMB_STATE_RECOVERY) { mcast_queue = &retrans_message_queue; sort_queue = &recovery_sort_queue; reset_token_retransmit_timeout (); // REVIEWED } else { mcast_queue = &new_message_queue; sort_queue = ®ular_sort_queue; } for (fcc_mcast_current = 0; fcc_mcast_current < fcc_mcasts_allowed; fcc_mcast_current++) { if (queue_is_empty (mcast_queue)) { break; } message_item = (struct message_item *)queue_item_get (mcast_queue); /* preincrement required by algo */ if (old_ring_state_saved && (memb_state == MEMB_STATE_GATHER || memb_state == MEMB_STATE_COMMIT)) { totemsrp_log_printf (totemsrp_log_level_debug, "not multicasting at seqno is %d\n", token->seq); return (0); } message_item->mcast->seq = ++token->seq; message_item->mcast->this_seqno = global_seqno++; /* * Build IO vector */ memset (&sort_queue_item, 0, sizeof (struct sort_queue_item)); sort_queue_item.iovec[0].iov_base = message_item->mcast; sort_queue_item.iovec[0].iov_len = sizeof (struct mcast); mcast = sort_queue_item.iovec[0].iov_base; memcpy (&sort_queue_item.iovec[1], message_item->iovec, message_item->iov_len * sizeof (struct iovec)); memcpy (&mcast->ring_id, &my_ring_id, sizeof (struct memb_ring_id)); sort_queue_item.iov_len = message_item->iov_len + 1; assert (sort_queue_item.iov_len < 16); /* * Add message to retransmit queue */ sq_item_add (sort_queue, &sort_queue_item, message_item->mcast->seq); // XXX printf ("ORIG [%s.%d-%d]\n", inet_ntoa (message_item->mcast->source), // XXX message_item->mcast->seq, message_item->mcast->this_seqno); /* * Delete item from pending queue */ queue_item_remove (mcast_queue); /* * Encrypt and digest the message */ encrypt_and_sign (sort_queue_item.iovec, sort_queue_item.iov_len); /* * Build multicast message */ msg_mcast.msg_name = &sockaddr_in_mcast; msg_mcast.msg_namelen = sizeof (struct sockaddr_in); msg_mcast.msg_iov = &iov_encrypted; msg_mcast.msg_iovlen = 1; msg_mcast.msg_control = 0; msg_mcast.msg_controllen = 0; msg_mcast.msg_flags = 0; /* * Multicast message * An error here is recovered by the multicast algorithm */ res = sendmsg (totemsrp_sockets[0].mcast, &msg_mcast, MSG_NOSIGNAL | MSG_DONTWAIT); iov_encrypted.iov_len = PACKET_SIZE_MAX; if (res > 0) { stats_sent += res; } } assert (fcc_mcast_current < 100); /* * If messages mcasted, deliver any new messages to totemg */ if (fcc_mcast_current) { my_do_delivery = 1; } my_high_seq_received = token->seq; update_aru (); /* * Return 1 if more messages are available for single node clusters */ return (fcc_mcast_current); } /* * Remulticasts messages in orf_token's retransmit list (requires orf_token) * Modify's orf_token's rtr to include retransmits required by this process */ static int orf_token_rtr ( struct orf_token *orf_token, int *fcc_allowed) { int res; int i, j; int found; int total_entries; struct sq *sort_queue; struct rtr_item *rtr_list; if (memb_state == MEMB_STATE_RECOVERY) { sort_queue = &recovery_sort_queue; } else { sort_queue = ®ular_sort_queue; } rtr_list = &orf_token->rtr_list[0]; if (orf_token->rtr_list_entries) { totemsrp_log_printf (totemsrp_log_level_debug, "Retransmit List %d\n", orf_token->rtr_list_entries); for (i = 0; i < orf_token->rtr_list_entries; i++) { totemsrp_log_printf (totemsrp_log_level_debug, "%d ", rtr_list[i].seq); } totemsrp_log_printf (totemsrp_log_level_debug, "\n"); } total_entries = orf_token->rtr_list_entries; /* * Retransmit messages on orf_token's RTR list from RTR queue */ for (fcc_remcast_current = 0, i = 0; fcc_remcast_current <= *fcc_allowed && i < orf_token->rtr_list_entries;) { /* * If this retransmit request isn't from this configuration, * try next rtr entry */ if (memcmp (&rtr_list[i].ring_id, &my_ring_id, sizeof (struct memb_ring_id)) != 0) { i += 1; continue; } assert (rtr_list[i].seq > 0); res = orf_token_remcast (rtr_list[i].seq); if (res == 0) { /* * Multicasted message, so no need to copy to new retransmit list */ orf_token->rtr_list_entries -= 1; assert (orf_token->rtr_list_entries >= 0); memmove (&rtr_list[i], &rtr_list[i + 1], sizeof (struct rtr_item) * (orf_token->rtr_list_entries)); fcc_remcast_current++; stats_remcasts++; } else { i += 1; } } *fcc_allowed = *fcc_allowed - fcc_remcast_current - 1; #ifdef COMPILE_OUT for (i = 0; i < orf_token->rtr_list_entries; i++) { assert (rtr_list_old[index_old].seq != -1); } #endif /* * Add messages to retransmit to RTR list * but only retry if there is room in the retransmit list */ for (i = my_aru + 1; orf_token->rtr_list_entries < RETRANSMIT_ENTRIES_MAX && i <= my_high_seq_received; i++) { /* * Find if a message is missing from this processor */ res = sq_item_inuse (sort_queue, i); if (res == 0) { /* * Determine if missing message is already in retransmit list */ found = 0; for (j = 0; j < orf_token->rtr_list_entries; j++) { if (i == rtr_list[j].seq) { found = 1; } } if (found == 0) { /* * Missing message not found in current retransmit list so add it */ memcpy (&rtr_list[orf_token->rtr_list_entries].ring_id, &my_ring_id, sizeof (struct memb_ring_id)); rtr_list[orf_token->rtr_list_entries].seq = i; orf_token->rtr_list_entries++; } } } return (fcc_remcast_current); } void token_retransmit (void) { struct iovec iovec; struct msghdr msg_orf_token; int res; iovec.iov_base = orf_token_retransmit; iovec.iov_len = orf_token_retransmit_size; msg_orf_token.msg_name = &next_memb; msg_orf_token.msg_namelen = sizeof (struct sockaddr_in); msg_orf_token.msg_iov = &iovec; msg_orf_token.msg_iovlen = 1; msg_orf_token.msg_control = 0; msg_orf_token.msg_controllen = 0; msg_orf_token.msg_flags = 0; res = sendmsg (totemsrp_sockets[0].token, &msg_orf_token, MSG_NOSIGNAL); } /* * Retransmit the regular token if no mcast or token has * been received in retransmit token period retransmit * the token to the next processor */ void timer_function_token_retransmit_timeout (void *data) { switch (memb_state) { case MEMB_STATE_GATHER: break; case MEMB_STATE_COMMIT: break; case MEMB_STATE_OPERATIONAL: case MEMB_STATE_RECOVERY: token_retransmit (); reset_token_retransmit_timeout (); // REVIEWED break; } } void timer_function_token_hold_retransmit_timeout (void *data) { switch (memb_state) { case MEMB_STATE_GATHER: break; case MEMB_STATE_COMMIT: break; case MEMB_STATE_OPERATIONAL: case MEMB_STATE_RECOVERY: token_retransmit (); break; } } void timer_function_merge_detect_timeout(void *data) { my_merge_detect_timeout_outstanding = 0; switch (memb_state) { case MEMB_STATE_OPERATIONAL: if (my_ring_id.rep.s_addr == my_id.sin_addr.s_addr) { memb_merge_detect_transmit (); } break; case MEMB_STATE_GATHER: case MEMB_STATE_COMMIT: case MEMB_STATE_RECOVERY: break; } } /* * Send orf_token to next member (requires orf_token) */ static int token_send ( struct orf_token *orf_token, int forward_token) { struct msghdr msg_orf_token; struct iovec iovec; int res; iovec.iov_base = (char *)orf_token; iovec.iov_len = sizeof (struct orf_token) + (orf_token->rtr_list_entries * sizeof (struct rtr_item)); encrypt_and_sign (&iovec, 1); /* * Keep an encrypted copy in case the token retransmit timer expires */ memcpy (orf_token_retransmit, iov_encrypted.iov_base, iov_encrypted.iov_len); orf_token_retransmit_size = iov_encrypted.iov_len; /* * IF the user doesn't want the token forwarded, then dont send * it but keep an encrypted copy for the retransmit timeout */ if (forward_token == 0) { return (0); } /* * Send the message */ msg_orf_token.msg_name = &next_memb; msg_orf_token.msg_namelen = sizeof (struct sockaddr_in); msg_orf_token.msg_iov = &iov_encrypted; msg_orf_token.msg_iovlen = 1; msg_orf_token.msg_control = 0; msg_orf_token.msg_controllen = 0; msg_orf_token.msg_flags = 0; res = sendmsg (totemsrp_sockets[0].token, &msg_orf_token, MSG_NOSIGNAL); if (res == -1) { totemsrp_log_printf (totemsrp_log_level_notice, "Couldn't send token to addr %s %s %d\n", inet_ntoa (next_memb.sin_addr), strerror (errno), totemsrp_sockets[0].token); } /* * res not used here errors are handled by algorithm */ if (res > 0) { stats_sent += res; } return (res); } static int token_hold_cancel_send (void) { struct token_hold_cancel token_hold_cancel; struct iovec iov; struct msghdr msghdr; /* * Only cancel if the token is currently held */ if (my_token_held == 0) { return (0); } my_token_held = 0; /* * Build message */ token_hold_cancel.header.type = MESSAGE_TYPE_TOKEN_HOLD_CANCEL; token_hold_cancel.header.endian_detector = ENDIAN_LOCAL; memcpy (&token_hold_cancel.ring_id, &my_ring_id, sizeof (struct memb_ring_id)); iov.iov_base = &token_hold_cancel; iov.iov_len = sizeof (struct token_hold_cancel); encrypt_and_sign (&iov, 1); /* * Build multicast message */ msghdr.msg_name = (caddr_t)&sockaddr_in_mcast; msghdr.msg_namelen = sizeof (struct sockaddr_in); msghdr.msg_iov = &iov_encrypted; msghdr.msg_iovlen = 1; msghdr.msg_control = 0; msghdr.msg_controllen = 0; msghdr.msg_flags = 0; /* * Multicast message */ sendmsg (totemsrp_sockets[0].mcast, &msghdr, MSG_NOSIGNAL | MSG_DONTWAIT); return (0); } int orf_token_send_initial (void) { struct orf_token orf_token; int res; orf_token.header.type = MESSAGE_TYPE_ORF_TOKEN; orf_token.header.endian_detector = ENDIAN_LOCAL; orf_token.header.encapsulated = 0; orf_token.seq = 0; orf_token.token_seq = 0; orf_token.retrans_flg = 1; my_set_retrans_flg = 1; /* if (queue_is_empty (&retrans_message_queue) == 1) { orf_token.retrans_flg = 0; } else { orf_token.retrans_flg = 1; my_set_retrans_flg = 1; } */ orf_token.aru = 0; // orf_token.aru_addr.s_addr = 0;//my_id.sin_addr.s_addr; orf_token.aru_addr.s_addr = my_id.sin_addr.s_addr; memcpy (&orf_token.ring_id, &my_ring_id, sizeof (struct memb_ring_id)); orf_token.fcc = 0; orf_token.rtr_list_entries = 0; res = token_send (&orf_token, 1); return (res); } static void memb_state_commit_token_update (struct memb_commit_token *memb_commit_token) { int memb_index_this; memb_index_this = (memb_commit_token->memb_index + 1) % memb_commit_token->addr_entries; memcpy (&memb_commit_token->memb_list[memb_index_this].ring_id, &my_old_ring_id, sizeof (struct memb_ring_id)); assert (my_old_ring_id.rep.s_addr != 0); memb_commit_token->memb_list[memb_index_this].aru = old_ring_state_aru; /* * TODO high delivered is really my_aru, but with safe this * could change? */ memb_commit_token->memb_list[memb_index_this].high_delivered = my_high_delivered; memb_commit_token->memb_list[memb_index_this].received_flg = my_received_flg; } static int memb_state_commit_token_send (struct memb_commit_token *memb_commit_token) { struct msghdr msghdr; struct iovec iovec; int res; int memb_index_this; int memb_index_next; memb_commit_token->token_seq++; memb_index_this = (memb_commit_token->memb_index + 1) % memb_commit_token->addr_entries; memb_index_next = (memb_index_this + 1) % memb_commit_token->addr_entries; memb_commit_token->memb_index = memb_index_this; iovec.iov_base = memb_commit_token; iovec.iov_len = sizeof (struct memb_commit_token); encrypt_and_sign (&iovec, 1); next_memb.sin_addr.s_addr = memb_commit_token->addr[memb_index_next].s_addr; next_memb.sin_family = AF_INET; next_memb.sin_port = sockaddr_in_mcast.sin_port; msghdr.msg_name = &next_memb; msghdr.msg_namelen = sizeof (struct sockaddr_in); msghdr.msg_iov = &iov_encrypted; msghdr.msg_iovlen = 1; msghdr.msg_control = 0; msghdr.msg_controllen = 0; msghdr.msg_flags = 0; res = sendmsg (totemsrp_sockets[0].token, &msghdr, MSG_NOSIGNAL | MSG_DONTWAIT); return (res); } int memb_lowest_in_config (void) { struct in_addr token_memb[PROCESSOR_COUNT_MAX]; int token_memb_entries = 0; struct in_addr lowest_addr; int i; lowest_addr.s_addr = 0xFFFFFFFF; memb_set_subtract (token_memb, &token_memb_entries, my_proc_list, my_proc_list_entries, my_failed_list, my_failed_list_entries); /* * find representative by searching for smallest identifier */ for (i = 0; i < token_memb_entries; i++) { if (lowest_addr.s_addr > token_memb[i].s_addr) { lowest_addr.s_addr = token_memb[i].s_addr; } } return (my_id.sin_addr.s_addr == lowest_addr.s_addr); } static void memb_state_commit_token_create (struct memb_commit_token *commit_token) { struct in_addr token_memb[PROCESSOR_COUNT_MAX]; int token_memb_entries = 0; totemsrp_log_printf (totemsrp_log_level_notice, "Creating commit token because I am the rep.\n"); memb_set_subtract (token_memb, &token_memb_entries, my_proc_list, my_proc_list_entries, my_failed_list, my_failed_list_entries); memset (commit_token, 0, sizeof (struct memb_commit_token)); commit_token->header.type = MESSAGE_TYPE_MEMB_COMMIT_TOKEN; commit_token->header.endian_detector = ENDIAN_LOCAL; commit_token->header.encapsulated = 0; commit_token->ring_id.rep.s_addr = my_id.sin_addr.s_addr; commit_token->ring_id.seq = token_ring_id_seq + 4; qsort (token_memb, token_memb_entries, sizeof (struct in_addr), in_addr_compare); memcpy (commit_token->addr, token_memb, token_memb_entries * sizeof (struct in_addr)); memset (commit_token->memb_list, 0, sizeof (struct memb_commit_token_memb_entry) * PROCESSOR_COUNT_MAX); commit_token->memb_index = token_memb_entries - 1; commit_token->addr_entries = token_memb_entries; } int memb_join_message_send (void) { struct msghdr msghdr; struct iovec iovec; struct memb_join memb_join; int res; memb_join.header.type = MESSAGE_TYPE_MEMB_JOIN; memb_join.header.endian_detector = ENDIAN_LOCAL; memb_join.header.encapsulated = 0; memb_join.ring_seq = my_ring_id.seq; memcpy (memb_join.proc_list, my_proc_list, my_proc_list_entries * sizeof (struct in_addr)); memb_join.proc_list_entries = my_proc_list_entries; memcpy (memb_join.failed_list, my_failed_list, my_failed_list_entries * sizeof (struct in_addr)); memb_join.failed_list_entries = my_failed_list_entries; iovec.iov_base = &memb_join; iovec.iov_len = sizeof (struct memb_join); encrypt_and_sign (&iovec, 1); msghdr.msg_name = &sockaddr_in_mcast; msghdr.msg_namelen = sizeof (struct sockaddr_in); msghdr.msg_iov = &iov_encrypted; msghdr.msg_iovlen = 1; msghdr.msg_control = 0; msghdr.msg_controllen = 0; msghdr.msg_flags = 0; res = sendmsg (totemsrp_sockets[0].mcast, &msghdr, MSG_NOSIGNAL | MSG_DONTWAIT); return (res); } static int memb_merge_detect_transmit (void) { struct msghdr msghdr; struct iovec iovec; struct memb_merge_detect memb_merge_detect; int res; memb_merge_detect.header.type = MESSAGE_TYPE_MEMB_MERGE_DETECT; memb_merge_detect.header.endian_detector = ENDIAN_LOCAL; memb_merge_detect.header.encapsulated = 0; memcpy (&memb_merge_detect.ring_id, &my_ring_id, sizeof (struct memb_ring_id)); iovec.iov_base = &memb_merge_detect; iovec.iov_len = sizeof (struct memb_merge_detect); encrypt_and_sign (&iovec, 1); msghdr.msg_name = &sockaddr_in_mcast; msghdr.msg_namelen = sizeof (struct sockaddr_in); msghdr.msg_iov = &iov_encrypted; msghdr.msg_iovlen = 1; msghdr.msg_control = 0; msghdr.msg_controllen = 0; msghdr.msg_flags = 0; res = sendmsg (totemsrp_sockets[0].mcast, &msghdr, MSG_NOSIGNAL | MSG_DONTWAIT); return (res); } static void memb_ring_id_create_or_load ( struct memb_ring_id *memb_ring_id) { int fd; int res; char filename[256]; sprintf (filename, "/tmp/ringid_%s", inet_ntoa (my_id.sin_addr)); fd = open (filename, O_RDONLY, 0777); if (fd > 0) { res = read (fd, &memb_ring_id->seq, sizeof (unsigned long long)); assert (res == sizeof (unsigned long long)); close (fd); } else if (fd == -1 && errno == ENOENT) { memb_ring_id->seq = 0; umask(0); fd = open (filename, O_CREAT|O_RDWR, 0777); if (fd == -1) { printf ("couldn't create file %d %s\n", fd, strerror(errno)); } res = write (fd, &memb_ring_id->seq, sizeof (unsigned long long)); assert (res == sizeof (unsigned long long)); close (fd); } else { totemsrp_log_printf (totemsrp_log_level_warning, "Couldn't open %s %s\n", filename, strerror (errno)); } memb_ring_id->rep.s_addr = my_id.sin_addr.s_addr; assert (memb_ring_id->rep.s_addr); token_ring_id_seq = memb_ring_id->seq; } static void memb_ring_id_store ( struct memb_commit_token *commit_token) { char filename[256]; int fd; int res; sprintf (filename, "/tmp/ringid_%s", inet_ntoa (my_id.sin_addr)); fd = open (filename, O_WRONLY, 0777); if (fd == -1) { fd = open (filename, O_CREAT|O_RDWR, 0777); } if (fd == -1) { totemsrp_log_printf (totemsrp_log_level_warning, "Couldn't store new ring id %llx to stable storage (%s)\n", commit_token->ring_id.seq, strerror (errno)); assert (0); return; } totemsrp_log_printf (totemsrp_log_level_notice, "Storing new sequence id for ring %d\n", commit_token->ring_id.seq); assert (fd > 0); res = write (fd, &commit_token->ring_id.seq, sizeof (unsigned long long)); assert (res == sizeof (unsigned long long)); close (fd); memcpy (&my_ring_id, &commit_token->ring_id, sizeof (struct memb_ring_id)); token_ring_id_seq = my_ring_id.seq; } void print_stats (void) { struct timeval tv_end; gettimeofday (&tv_end, NULL); totemsrp_log_printf (totemsrp_log_level_notice, "Bytes recv %d\n", stats_recv); totemsrp_log_printf (totemsrp_log_level_notice, "Bytes sent %d\n", stats_sent); totemsrp_log_printf (totemsrp_log_level_notice, "Messages delivered %d\n", stats_delv); totemsrp_log_printf (totemsrp_log_level_notice, "Re-Mcasts %d\n", stats_remcasts); totemsrp_log_printf (totemsrp_log_level_notice, "Tokens process %d\n", stats_orf_token); } int totemsrp_callback_token_create (void **handle_out, enum totem_callback_token_type type, int delete, int (*callback_fn) (enum totem_callback_token_type type, void *), void *data) { struct token_callback_instance *handle; handle = (struct token_callback_instance *)malloc (sizeof (struct token_callback_instance)); if (handle == 0) { return (-1); } *handle_out = (void *)handle; list_init (&handle->list); handle->callback_fn = callback_fn; handle->data = data; handle->callback_type = type; handle->delete = delete; switch (type) { case TOTEM_CALLBACK_TOKEN_RECEIVED: list_add (&handle->list, &token_callback_received_listhead); break; case TOTEM_CALLBACK_TOKEN_SENT: list_add (&handle->list, &token_callback_sent_listhead); break; } return (0); } void totemsrp_callback_token_destroy (void **handle_out) { struct token_callback_instance *h; if (*handle_out) { h = (struct token_callback_instance *)*handle_out; list_del (&h->list); free (h); h = NULL; *handle_out = 0; } } void totem_callback_token_type (void *handle) { struct token_callback_instance *token_callback_instance = (struct token_callback_instance *)handle; list_del (&token_callback_instance->list); free (token_callback_instance); } static void token_callbacks_execute (enum totem_callback_token_type type) { struct list_head *list; struct list_head *list_next; struct list_head *callback_listhead = 0; struct token_callback_instance *token_callback_instance; int res; int del; switch (type) { case TOTEM_CALLBACK_TOKEN_RECEIVED: callback_listhead = &token_callback_received_listhead; break; case TOTEM_CALLBACK_TOKEN_SENT: callback_listhead = &token_callback_sent_listhead; break; default: assert (0); } for (list = callback_listhead->next; list != callback_listhead; list = list_next) { token_callback_instance = list_entry (list, struct token_callback_instance, list); list_next = list->next; del = token_callback_instance->delete; if (del == 1) { list_del (list); } res = token_callback_instance->callback_fn ( token_callback_instance->callback_type, token_callback_instance->data); /* * This callback failed to execute, try it again on the next token */ if (res == -1 && del == 1) { list_add (list, callback_listhead); } else if (del) { free (token_callback_instance); } } } /* * Message Handlers */ int my_last_seq = 0; struct timeval tv_old; /* * message handler called when TOKEN message type received */ static int message_handler_orf_token ( struct sockaddr_in *system_from, struct iovec *iovec, int iov_len, int bytes_received, int endian_conversion_needed) { char token_storage[1500]; char token_convert[1500]; struct orf_token *token; int prio = UINT_MAX; struct pollfd ufd; int nfds; struct orf_token *token_ref = (struct orf_token *)iovec->iov_base; int transmits_allowed; int forward_token; int mcasted; int last_aru; int low_water; #ifdef GIVEINFO struct timeval tv_current; struct timeval tv_diff; gettimeofday (&tv_current, NULL); timersub (&tv_current, &tv_old, &tv_diff); memcpy (&tv_old, &tv_current, sizeof (struct timeval)); if ((((float)tv_diff.tv_usec) / 100.0) > 5.0) { printf ("OTHERS %0.4f ms\n", ((float)tv_diff.tv_usec) / 100.0); } #endif my_do_delivery = 0; #ifdef RANDOM_DROP if (random () % 100 < 10) { return (0); } #endif /* * Handle merge detection timeout */ if (token_ref->seq == my_last_seq) { start_merge_detect_timeout (); my_seq_unchanged += 1; } else { cancel_merge_detect_timeout (); cancel_token_hold_retransmit_timeout (); my_seq_unchanged = 0; } my_last_seq = token_ref->seq; assert (bytes_received >= sizeof (struct orf_token)); // assert (bytes_received == sizeof (struct orf_token) + // (sizeof (struct rtr_item) * token_ref->rtr_list_entries); /* * Make copy of token and retransmit list in case we have * to flush incoming messages from the kernel queue */ token = (struct orf_token *)token_storage; memcpy (token, iovec->iov_base, sizeof (struct orf_token)); memcpy (&token->rtr_list[0], iovec->iov_base + sizeof (struct orf_token), sizeof (struct rtr_item) * RETRANSMIT_ENTRIES_MAX); if (endian_conversion_needed) { orf_token_endian_convert (token, (struct orf_token *)token_convert); token = (struct orf_token *)token_convert; } /* * flush incoming queue from kernel */ do { ufd.fd = totemsrp_sockets[0].mcast; ufd.events = POLLIN; nfds = poll (&ufd, 1, 0); if (nfds == 1 && ufd.revents & POLLIN) { totemsrp_iov_recv.iov_len = PACKET_SIZE_MAX; recv_handler (0, totemsrp_sockets[0].mcast, ufd.revents, 0, &prio); } } while (nfds == 1); /* * Determine if we should hold (in reality drop) the token */ my_token_held = 0; if (my_ring_id.rep.s_addr == my_id.sin_addr.s_addr && my_seq_unchanged > SEQNO_UNCHANGED_CONST) { my_token_held = 1; } else if (my_ring_id.rep.s_addr != my_id.sin_addr.s_addr && my_seq_unchanged >= SEQNO_UNCHANGED_CONST) { my_token_held = 1; } /* * Hold onto token when there is no activity on ring and * this processor is the ring rep */ forward_token = 1; if (my_ring_id.rep.s_addr == my_id.sin_addr.s_addr) { if (my_token_held) { forward_token = 0; } } token_callbacks_execute (TOTEM_CALLBACK_TOKEN_RECEIVED); switch (memb_state) { case MEMB_STATE_COMMIT: /* Discard token */ break; case MEMB_STATE_OPERATIONAL: messages_free (token->aru); case MEMB_STATE_GATHER: /* * DO NOT add break, we use different free mechanism in recovery state */ case MEMB_STATE_RECOVERY: last_aru = my_last_aru; my_last_aru = token->aru; /* * Discard tokens from another configuration */ if (memcmp (&token->ring_id, &my_ring_id, sizeof (struct memb_ring_id)) != 0) { return (0); /* discard token */ } /* * Discard retransmitted tokens */ if (my_token_seq >= token->token_seq) { reset_token_retransmit_timeout (); reset_token_timeout (); return (0); /* discard token */ } transmits_allowed = 30; mcasted = orf_token_rtr (token, &transmits_allowed); if ((last_aru + MISSING_MCAST_WINDOW) < token->seq) { transmits_allowed = 0; } mcasted = orf_token_mcast (token, transmits_allowed, system_from); if (my_aru < token->aru || my_id.sin_addr.s_addr == token->aru_addr.s_addr || token->aru_addr.s_addr == 0) { token->aru = my_aru; if (token->aru == token->seq) { token->aru_addr.s_addr = 0; } else { token->aru_addr.s_addr = my_id.sin_addr.s_addr; } } if (token->aru == last_aru && token->aru_addr.s_addr != 0) { my_aru_count += 1; } else { my_aru_count = 0; } if (my_aru_count > fail_to_recv_const && token->aru_addr.s_addr != my_id.sin_addr.s_addr) { printf ("FAILED TO RECEIVE\n"); // TODO if we fail to receive, it may be possible to end with a gather // state of proc == failed = 0 entries memb_set_merge (&token->aru_addr, 1, my_failed_list, &my_failed_list_entries); ring_state_restore (); memb_state_gather_enter (); } else { my_token_seq = token->token_seq; token->token_seq += 1; if (memb_state == MEMB_STATE_RECOVERY) { /* * my_aru == my_high_seq_received means this processor * has recovered all messages it can recover * (ie: its retrans queue is empty) */ low_water = my_aru; if (low_water > last_aru) { low_water = last_aru; } // TODO is this code right if (queue_is_empty (&retrans_message_queue) == 0 || low_water != my_high_seq_received) { if (token->retrans_flg == 0) { token->retrans_flg = 1; my_set_retrans_flg = 1; } } else if (token->retrans_flg == 1 && my_set_retrans_flg) { token->retrans_flg = 0; } totemsrp_log_printf (totemsrp_log_level_debug, "token retrans flag is %d my set retrans flag%d retrans queue empty %d count %d, low_water %d aru %d\n", token->retrans_flg, my_set_retrans_flg, queue_is_empty (&retrans_message_queue), my_retrans_flg_count, low_water, token->aru); if (token->retrans_flg == 0) { my_retrans_flg_count += 1; } else { my_retrans_flg_count = 0; } if (my_retrans_flg_count == 2) { my_install_seq = token->seq; } totemsrp_log_printf (totemsrp_log_level_debug, "install seq %d aru %d high seq received %d\n", my_install_seq, my_aru, my_high_seq_received); if (my_retrans_flg_count >= 2 && my_aru >= my_install_seq && my_received_flg == 0) { my_received_flg = 1; my_deliver_memb_entries = my_trans_memb_entries; memcpy (my_deliver_memb_list, my_trans_memb_list, sizeof (struct in_addr) * my_trans_memb_entries); } if (my_retrans_flg_count >= 3 && token->aru >= my_install_seq) { my_rotation_counter += 1; } else { my_rotation_counter = 0; } if (my_rotation_counter == 2) { totemsrp_log_printf (totemsrp_log_level_debug, "retrans flag count %d token aru %d install seq %d aru %d %d\n", my_retrans_flg_count, token->aru, my_install_seq, my_aru, token->seq); memb_state_operational_enter (); my_rotation_counter = 0; my_retrans_flg_count = 0; } } token_send (token, forward_token); #ifdef GIVEINFO gettimeofday (&tv_current, NULL); timersub (&tv_current, &tv_old, &tv_diff); memcpy (&tv_old, &tv_current, sizeof (struct timeval)); if ((((float)tv_diff.tv_usec) / 100.0) > 5.0) { printf ("I held %0.4f ms\n", ((float)tv_diff.tv_usec) / 100.0); } #endif if (my_do_delivery) { if (memb_state == MEMB_STATE_OPERATIONAL) { messages_deliver_to_app (0, my_high_seq_received); } } /* * Deliver messages after token has been transmitted * to improve performance */ reset_token_timeout (); // REVIEWED reset_token_retransmit_timeout (); // REVIEWED if (my_id.sin_addr.s_addr == my_ring_id.rep.s_addr && my_token_held == 1) { start_token_hold_retransmit_timeout (); } token_callbacks_execute (TOTEM_CALLBACK_TOKEN_SENT); } break; } return (0); } static void messages_deliver_to_app (int skip, int end_point) { struct sort_queue_item *sort_queue_item_p; int i; int res; struct mcast *mcast; totemsrp_log_printf (totemsrp_log_level_debug, "Delivering %d to %d\n", my_high_delivered + 1, end_point); /* * Deliver messages in order from rtr queue to pending delivery queue */ for (i = my_high_delivered + 1; i <= end_point; i++) { void *ptr; res = sq_item_get (®ular_sort_queue, i, &ptr); if (res != 0 && skip) { printf ("-skipping %d-\n", i); my_high_delivered = i; continue; } /* * If hole, stop assembly */ if (res != 0) { break; } sort_queue_item_p = ptr; mcast = sort_queue_item_p->iovec[0].iov_base; assert (mcast != (struct mcast *)0xdeadbeef); // XXX printf ("[%s.%d-%d]\n", inet_ntoa (mcast->source), // XXX mcast->seq, mcast->this_seqno); /* * Skip messages not originated in my_deliver_memb */ if (skip && memb_set_subset (&mcast->source, 1, my_deliver_memb_list, my_deliver_memb_entries) == 0) { printf ("-skipping %d - wrong ip", i); my_high_delivered = i; continue; } my_high_delivered = i; /* * Message found */ totemsrp_log_printf (totemsrp_log_level_debug, "Delivering MCAST message with seq %d to pending delivery queue\n", mcast->seq); /* * Message is locally originated multicast */ if (sort_queue_item_p->iov_len > 1 && sort_queue_item_p->iovec[0].iov_len == sizeof (struct mcast)) { totemsrp_deliver_fn ( mcast->source, &sort_queue_item_p->iovec[1], sort_queue_item_p->iov_len - 1, mcast->header.endian_detector != ENDIAN_LOCAL); } else { sort_queue_item_p->iovec[0].iov_len -= sizeof (struct mcast); sort_queue_item_p->iovec[0].iov_base += sizeof (struct mcast); totemsrp_deliver_fn ( mcast->source, sort_queue_item_p->iovec, sort_queue_item_p->iov_len, mcast->header.endian_detector != ENDIAN_LOCAL); sort_queue_item_p->iovec[0].iov_len += sizeof (struct mcast); sort_queue_item_p->iovec[0].iov_base -= sizeof (struct mcast); } stats_delv += 1; } my_received_flg = 0; if (my_aru == my_high_seq_received) { // totemsrp_log_printf (totemsrp_log_level_debug, // "setting received flag to TRUE %d %d\n", // my_aru, my_high_seq_received); my_received_flg = 1; } } /* * recv message handler called when MCAST message type received */ static int message_handler_mcast ( struct sockaddr_in *system_from, struct iovec *iovec, int iov_len, int bytes_received, int endian_conversion_needed) { struct sort_queue_item sort_queue_item; struct sq *sort_queue; struct mcast mcast_header; if (endian_conversion_needed) { mcast_endian_convert (iovec[0].iov_base, &mcast_header); } else { memcpy (&mcast_header, iovec[0].iov_base, sizeof (struct mcast)); } if (mcast_header.header.encapsulated == 1) { sort_queue = &recovery_sort_queue; } else { sort_queue = ®ular_sort_queue; } assert (bytes_received < PACKET_SIZE_MAX); #ifdef RANDOM_DROP if (random()%100 < 50) { return (0); } #endif if (system_from->sin_addr.s_addr != my_id.sin_addr.s_addr) { cancel_token_retransmit_timeout (); } /* * If the message is foreign execute the switch below */ if (memcmp (&my_ring_id, &mcast_header.ring_id, sizeof (struct memb_ring_id)) != 0) { switch (memb_state) { case MEMB_STATE_OPERATIONAL: memb_set_merge (&system_from->sin_addr, 1, my_proc_list, &my_proc_list_entries); memb_state_gather_enter (); break; case MEMB_STATE_GATHER: if (!memb_set_subset (&system_from->sin_addr, 1, my_proc_list, my_proc_list_entries)) { memb_set_merge (&system_from->sin_addr, 1, my_proc_list, &my_proc_list_entries); memb_state_gather_enter (); return (0); } break; case MEMB_STATE_COMMIT: /* discard message */ break; case MEMB_STATE_RECOVERY: /* discard message */ break; } return (0); } totemsrp_log_printf (totemsrp_log_level_debug, "Received ringid(%s:%lld) seq %d\n", inet_ntoa (mcast_header.ring_id.rep), mcast_header.ring_id.seq, mcast_header.seq); /* * Add mcast message to rtr queue if not already in rtr queue * otherwise free io vectors */ if (bytes_received > 0 && bytes_received < PACKET_SIZE_MAX && my_aru < mcast_header.seq && sq_item_inuse (sort_queue, mcast_header.seq) == 0) { /* * Allocate new multicast memory block */ // TODO LEAK sort_queue_item.iovec[0].iov_base = malloc (bytes_received); if (sort_queue_item.iovec[0].iov_base == 0) { return (-1); /* error here is corrected by the algorithm */ } memcpy (sort_queue_item.iovec[0].iov_base, iovec[0].iov_base, bytes_received); sort_queue_item.iovec[0].iov_len = bytes_received; assert (sort_queue_item.iovec[0].iov_len > 0); assert (sort_queue_item.iovec[0].iov_len < PACKET_SIZE_MAX); sort_queue_item.iov_len = 1; if (mcast_header.seq > my_high_seq_received) { my_high_seq_received = mcast_header.seq; } sq_item_add (sort_queue, &sort_queue_item, mcast_header.seq); } // update_aru (); if (my_token_held) { my_do_delivery = 1; } else { if (memb_state == MEMB_STATE_OPERATIONAL) { update_aru (); messages_deliver_to_app (0, my_high_seq_received); } } /* TODO remove from retrans message queue for old ring in recovery state */ return (0); } static int message_handler_memb_merge_detect ( struct sockaddr_in *system_from, struct iovec *iovec, int iov_len, int bytes_received, int endian_conversion_needed) { struct memb_merge_detect *memb_merge_detect = iovec[0].iov_base; /* * do nothing if this is a merge detect from this configuration */ if (memcmp (&my_ring_id, &memb_merge_detect->ring_id, sizeof (struct memb_ring_id)) == 0) { return (0); } /* * Execute merge operation */ switch (memb_state) { case MEMB_STATE_OPERATIONAL: memb_set_merge (&system_from->sin_addr, 1, my_proc_list, &my_proc_list_entries); memb_state_gather_enter (); break; case MEMB_STATE_GATHER: if (!memb_set_subset (&system_from->sin_addr, 1, my_proc_list, my_proc_list_entries)) { memb_set_merge (&system_from->sin_addr, 1, my_proc_list, &my_proc_list_entries); memb_state_gather_enter (); return (0); } break; case MEMB_STATE_COMMIT: /* do nothing in commit */ break; case MEMB_STATE_RECOVERY: /* do nothing in recovery */ break; } return (0); } int memb_join_process (struct memb_join *memb_join, struct sockaddr_in *system_from) { struct memb_commit_token my_commit_token; if (memb_set_equal (memb_join->proc_list, memb_join->proc_list_entries, my_proc_list, my_proc_list_entries) && memb_set_equal (memb_join->failed_list, memb_join->failed_list_entries, my_failed_list, my_failed_list_entries)) { memb_consensus_set (&system_from->sin_addr); if (memb_consensus_agreed () && memb_lowest_in_config ()) { memb_state_commit_token_create (&my_commit_token); memb_state_commit_enter (&my_commit_token); } else { return (0); } } else if (memb_set_subset (memb_join->proc_list, memb_join->proc_list_entries, my_proc_list, my_proc_list_entries) && memb_set_subset (memb_join->failed_list, memb_join->failed_list_entries, my_failed_list, my_failed_list_entries)) { return (0); } else if (memb_set_subset (&system_from->sin_addr, 1, my_failed_list, my_failed_list_entries)) { return (0); } else { memb_set_merge (memb_join->proc_list, memb_join->proc_list_entries, my_proc_list, &my_proc_list_entries); if (memb_set_subset (&my_id.sin_addr, 1, memb_join->failed_list, memb_join->failed_list_entries)) { memb_set_merge (&system_from->sin_addr, 1, my_failed_list, &my_failed_list_entries); } else { memb_set_merge (memb_join->failed_list, memb_join->failed_list_entries, my_failed_list, &my_failed_list_entries); } memb_state_gather_enter (); return (1); /* gather entered */ } return (0); /* gather not entered */ } static void memb_join_endian_convert (struct memb_join *in, struct memb_join *out) { int i; out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->proc_list_entries = swab32 (in->proc_list_entries); out->failed_list_entries = swab32 (in->failed_list_entries); out->ring_seq = swab64 (in->ring_seq); for (i = 0; i < out->proc_list_entries; i++) { out->proc_list[i].s_addr = in->proc_list[i].s_addr; } for (i = 0; i < out->failed_list_entries; i++) { out->failed_list[i].s_addr = in->failed_list[i].s_addr; } } static void memb_commit_token_endian_convert (struct memb_commit_token *in, struct memb_commit_token *out) { int i; out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->token_seq = swab32 (in->token_seq); out->ring_id.rep.s_addr = in->ring_id.rep.s_addr; out->ring_id.seq = swab64 (in->ring_id.seq); out->retrans_flg = swab32 (in->retrans_flg); out->memb_index = swab32 (in->memb_index); out->addr_entries = swab32 (in->addr_entries); for (i = 0; i < out->addr_entries; i++) { out->addr[i].s_addr = in->addr[i].s_addr; out->memb_list[i].ring_id.rep.s_addr = in->memb_list[i].ring_id.rep.s_addr; out->memb_list[i].ring_id.seq = swab64 (in->memb_list[i].ring_id.seq); out->memb_list[i].aru = swab32 (in->memb_list[i].aru); out->memb_list[i].high_delivered = swab32 (in->memb_list[i].high_delivered); out->memb_list[i].received_flg = swab32 (in->memb_list[i].received_flg); } } static void orf_token_endian_convert (struct orf_token *in, struct orf_token *out) { int i; out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->seq = swab32 (in->seq); out->token_seq = swab32 (in->token_seq); out->aru = swab32 (in->aru); out->ring_id.rep.s_addr = in->ring_id.rep.s_addr; out->ring_id.seq = swab64 (in->ring_id.seq); out->fcc = swab32 (in->fcc); out->retrans_flg = swab32 (in->retrans_flg); out->rtr_list_entries = swab32 (in->rtr_list_entries); for (i = 0; i < out->rtr_list_entries; i++) { out->rtr_list[i].ring_id.rep.s_addr = in->rtr_list[i].ring_id.rep.s_addr; out->rtr_list[i].ring_id.seq = swab64 (in->rtr_list[i].ring_id.seq); out->rtr_list[i].seq = swab32 (in->rtr_list[i].seq); } } static void mcast_endian_convert (struct mcast *in, struct mcast *out) { out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->seq = swab32 (in->seq); out->ring_id.rep.s_addr = in->ring_id.rep.s_addr; out->ring_id.seq = swab64 (in->ring_id.seq); out->source = in->source; out->guarantee = in->guarantee; } static int message_handler_memb_join ( struct sockaddr_in *system_from, struct iovec *iovec, int iov_len, int bytes_received, int endian_conversion_needed) { struct memb_join *memb_join; struct memb_join memb_join_convert; int gather_entered; if (endian_conversion_needed) { memb_join = &memb_join_convert; memb_join_endian_convert (iovec->iov_base, &memb_join_convert); } else { memb_join = (struct memb_join *)iovec->iov_base; } if (token_ring_id_seq < memb_join->ring_seq) { token_ring_id_seq = memb_join->ring_seq; } switch (memb_state) { case MEMB_STATE_OPERATIONAL: gather_entered = memb_join_process (memb_join, system_from); if (gather_entered == 0) { memb_state_gather_enter (); } break; case MEMB_STATE_GATHER: memb_join_process (memb_join, system_from); break; case MEMB_STATE_COMMIT: if (memb_set_subset (&system_from->sin_addr, 1, my_new_memb_list, my_new_memb_entries) && memb_join->ring_seq >= my_ring_id.seq) { memb_join_process (memb_join, system_from); memb_state_gather_enter (); } break; case MEMB_STATE_RECOVERY: if (memb_set_subset (&system_from->sin_addr, 1, my_new_memb_list, my_new_memb_entries) && memb_join->ring_seq >= my_ring_id.seq) { ring_state_restore (); memb_join_process (memb_join, system_from); memb_state_gather_enter (); } break; } return (0); } static int message_handler_memb_commit_token ( struct sockaddr_in *system_from, struct iovec *iovec, int iov_len, int bytes_received, int endian_conversion_needed) { struct memb_commit_token memb_commit_token_convert; struct memb_commit_token *memb_commit_token; struct in_addr sub[PROCESSOR_COUNT_MAX]; int sub_entries; if (endian_conversion_needed) { memb_commit_token = &memb_commit_token_convert; memb_commit_token_endian_convert (iovec->iov_base, memb_commit_token); } else { memb_commit_token = (struct memb_commit_token *)iovec->iov_base; } /* TODO do we need to check for a duplicate token? if (memb_commit_token->token_seq > 0 && my_token_seq >= memb_commit_token->token_seq) { printf ("already received commit token %d %d\n", memb_commit_token->token_seq, my_token_seq); return (0); } */ #ifdef RANDOM_DROP if (random()%100 < 10) { return (0); } #endif switch (memb_state) { case MEMB_STATE_OPERATIONAL: /* discard token */ break; case MEMB_STATE_GATHER: memb_set_subtract (sub, &sub_entries, my_proc_list, my_proc_list_entries, my_failed_list, my_failed_list_entries); if (memb_set_equal (memb_commit_token->addr, memb_commit_token->addr_entries, sub, sub_entries) && memb_commit_token->ring_id.seq > my_ring_id.seq) { memb_state_commit_enter (memb_commit_token); } break; case MEMB_STATE_COMMIT: if (memcmp (&memb_commit_token->ring_id, &my_ring_id, sizeof (struct memb_ring_id)) == 0) { // if (memb_commit_token->ring_id.seq == my_ring_id.seq) { memb_state_recovery_enter (memb_commit_token); } break; case MEMB_STATE_RECOVERY: totemsrp_log_printf (totemsrp_log_level_notice, "Sending initial ORF token\n"); if (my_id.sin_addr.s_addr == my_ring_id.rep.s_addr) { // TODO convert instead of initiate orf_token_send_initial (); reset_token_timeout (); // REVIEWED reset_token_retransmit_timeout (); // REVIEWED } break; } return (0); } static int message_handler_token_hold_cancel ( struct sockaddr_in *system_from, struct iovec *iovec, int iov_len, int bytes_received, int endian_conversion_needed) { struct token_hold_cancel *token_hold_cancel = (struct token_hold_cancel *)iovec->iov_base; if (memcmp (&token_hold_cancel->ring_id, &my_ring_id, sizeof (struct memb_ring_id)) == 0) { my_seq_unchanged = 0; if (my_ring_id.rep.s_addr == my_id.sin_addr.s_addr) { timer_function_token_retransmit_timeout (0); } } return (0); } static int recv_handler (poll_handle handle, int fd, int revents, void *data, unsigned int *prio) { struct msghdr msg_recv; struct message_header *message_header; struct sockaddr_in system_from; int res = 0; int bytes_received; *prio = UINT_MAX; /* * Receive datagram */ msg_recv.msg_name = &system_from; msg_recv.msg_namelen = sizeof (struct sockaddr_in); msg_recv.msg_iov = &totemsrp_iov_recv; msg_recv.msg_iovlen = 1; msg_recv.msg_control = 0; msg_recv.msg_controllen = 0; msg_recv.msg_flags = 0; bytes_received = recvmsg (fd, &msg_recv, MSG_NOSIGNAL | MSG_DONTWAIT); if (bytes_received == -1) { return (0); } else { stats_recv += bytes_received; } if (bytes_received < sizeof (struct message_header)) { totemsrp_log_printf (totemsrp_log_level_security, "Received message is too short... ignoring %d %d.\n", bytes_received); return (0); } message_header = (struct message_header *)msg_recv.msg_iov->iov_base; /* * Authenticate and if authenticated, decrypt datagram */ totemsrp_iov_recv.iov_len = bytes_received; res = authenticate_and_decrypt (&totemsrp_iov_recv); log_digest = 0; if (res == -1) { totemsrp_iov_recv.iov_len = PACKET_SIZE_MAX; return 0; } if (stats_tv_start.tv_usec == 0) { gettimeofday (&stats_tv_start, NULL); } /* * Handle incoming message */ message_header = (struct message_header *)msg_recv.msg_iov[0].iov_base; totemsrp_message_handlers.handler_functions[(int)message_header->type] ( &system_from, msg_recv.msg_iov, msg_recv.msg_iovlen, bytes_received, message_header->endian_detector != ENDIAN_LOCAL); totemsrp_iov_recv.iov_len = PACKET_SIZE_MAX; return (0); }