|
|
@@ -0,0 +1,2861 @@
|
|
|
+/*
|
|
|
+ * 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.
|
|
|
+ */
|
|
|
+
|
|
|
+/*
|
|
|
+ * This code implements the ring protocol specified in Yair Amir's PhD thesis:
|
|
|
+ * http://www.cs.jhu.edu/~yairamir/phd.ps) (ch4,5).
|
|
|
+ *
|
|
|
+ * Some changes have been made to the design to support things like fragmentation,
|
|
|
+ * multiple I/O queues, and other things.
|
|
|
+ */
|
|
|
+
|
|
|
+#include <assert.h>
|
|
|
+#include <sys/poll.h>
|
|
|
+#include <sys/mman.h>
|
|
|
+#include <sys/types.h>
|
|
|
+#include <sys/socket.h>
|
|
|
+#include <netdb.h>
|
|
|
+#include <sys/un.h>
|
|
|
+#include <sys/sysinfo.h>
|
|
|
+#include <sys/ioctl.h>
|
|
|
+#include <netinet/in.h>
|
|
|
+#include <arpa/inet.h>
|
|
|
+#include <linux/if.h>
|
|
|
+#include <linux/sockios.h>
|
|
|
+#include <unistd.h>
|
|
|
+#include <fcntl.h>
|
|
|
+#include <stdlib.h>
|
|
|
+#include <stdio.h>
|
|
|
+#include <errno.h>
|
|
|
+#include <signal.h>
|
|
|
+#include <sched.h>
|
|
|
+#include <time.h>
|
|
|
+#include <sys/time.h>
|
|
|
+
|
|
|
+#include "poll.h"
|
|
|
+#include "gmi.h"
|
|
|
+#include "../include/queue.h"
|
|
|
+#include "../include/sq.h"
|
|
|
+#include "print.h"
|
|
|
+
|
|
|
+extern struct sockaddr_in this_ip;
|
|
|
+
|
|
|
+#define LOCALHOST_IP inet_addr("127.0.0.1")
|
|
|
+#define QUEUE_PEND_DELV_SIZE_MAX ((MESSAGE_SIZE_MAX / 1472) + 1) * 2
|
|
|
+#define QUEUE_RTR_ITEMS_SIZE_MAX 512
|
|
|
+#define QUEUE_PEND_TRANS_SIZE_MAX ((MESSAGE_SIZE_MAX / 1472) + 1) * 500
|
|
|
+#define MAXIOVS 8
|
|
|
+#define RTR_TOKEN_SIZE_MAX 32
|
|
|
+#define MISSING_MCAST_WINDOW 64
|
|
|
+#define TIMEOUT_STATE_GATHER 300
|
|
|
+#define TIMEOUT_TOKEN 300
|
|
|
+#define TIMEOUT_STATE_COMMIT 300
|
|
|
+#define MAX_MEMBERS 16
|
|
|
+#define HOLE_LIST_MAX MISSING_MCAST_WINDOW
|
|
|
+#define PRIORITY_MAX 3
|
|
|
+
|
|
|
+int stats_sent = 0;
|
|
|
+int stats_recv = 0;
|
|
|
+int stats_delv = 0;
|
|
|
+int stats_remcasts = 0;
|
|
|
+int stats_orf_token = 0;
|
|
|
+int stats_form_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_ATTEMPT_JOIN = 2, /* membership join attempt message */
|
|
|
+ MESSAGE_TYPE_MEMB_JOIN = 3, /* membership join message */
|
|
|
+ MESSAGE_TYPE_MEMB_FORM_TOKEN = 4 /* membership FORM token */
|
|
|
+};
|
|
|
+
|
|
|
+/*
|
|
|
+ * In-order pending transmit queue
|
|
|
+ */
|
|
|
+struct queue queues_pend_trans[PRIORITY_MAX];
|
|
|
+
|
|
|
+/*
|
|
|
+ * In-order pending delivery queue
|
|
|
+ */
|
|
|
+struct pend_delv {
|
|
|
+ struct in_addr ip;
|
|
|
+ int seqid;
|
|
|
+ int first_delivery;
|
|
|
+ struct queue queue;
|
|
|
+};
|
|
|
+
|
|
|
+struct pend_delv queues_pend_delv[MAX_MEMBERS];
|
|
|
+
|
|
|
+/*
|
|
|
+ * Sorted delivery/retransmit queue
|
|
|
+ */
|
|
|
+struct sq queue_rtr_items;
|
|
|
+
|
|
|
+/*
|
|
|
+ * Multicast address
|
|
|
+ */
|
|
|
+struct sockaddr_in sockaddr_in_mcast;
|
|
|
+
|
|
|
+/*
|
|
|
+ * Multicast file descriptor
|
|
|
+ */
|
|
|
+int gmi_fd;
|
|
|
+
|
|
|
+/*
|
|
|
+ * Received up to and including
|
|
|
+ */
|
|
|
+int gmi_arut = 0;
|
|
|
+
|
|
|
+/*
|
|
|
+ * Delivered up to and including
|
|
|
+ */
|
|
|
+int gmi_adut = 0;
|
|
|
+
|
|
|
+int gmi_adut_old = 0;
|
|
|
+
|
|
|
+int gmi_original_arut = 0;
|
|
|
+
|
|
|
+int gmi_highest_seq = 0;
|
|
|
+
|
|
|
+int gmi_highest_seq_old = 0;
|
|
|
+
|
|
|
+int gmi_barrier_seq = 0;
|
|
|
+
|
|
|
+int gmi_last_seqid = 0;
|
|
|
+
|
|
|
+int gmi_fragment = 0;
|
|
|
+
|
|
|
+int gmi_pend_queue_priority = 0;
|
|
|
+
|
|
|
+/*
|
|
|
+ * Timers
|
|
|
+ */
|
|
|
+poll_timer_handle timer_orf_token_timeout = 0;
|
|
|
+
|
|
|
+poll_timer_handle timer_form_token_timeout = 0;
|
|
|
+
|
|
|
+poll_timer_handle timer_memb_state_gather_timeout = 0;
|
|
|
+
|
|
|
+poll_timer_handle timer_memb_state_commit_timeout = 0;
|
|
|
+
|
|
|
+poll_timer_handle timer_single_member = 0;
|
|
|
+
|
|
|
+/*
|
|
|
+ * Function called when new message received
|
|
|
+ */
|
|
|
+int (*gmi_recv) (char *group, struct iovec *iovec, int iov_len);
|
|
|
+
|
|
|
+struct message_header {
|
|
|
+ int type;
|
|
|
+ int seqid;
|
|
|
+};
|
|
|
+
|
|
|
+struct memb_conf_id {
|
|
|
+ struct in_addr rep;
|
|
|
+ struct timeval tv;
|
|
|
+};
|
|
|
+
|
|
|
+struct mcast {
|
|
|
+ struct message_header header;
|
|
|
+ char priority;
|
|
|
+ struct memb_conf_id memb_conf_id;
|
|
|
+ short packet_number;
|
|
|
+ short packet_count;
|
|
|
+ int packet_seq;
|
|
|
+ struct in_addr source;
|
|
|
+ struct gmi_groupname groupname;
|
|
|
+};
|
|
|
+
|
|
|
+struct rtr_item {
|
|
|
+ struct memb_conf_id conf_id;
|
|
|
+ int seqid;
|
|
|
+};
|
|
|
+
|
|
|
+struct orf_token {
|
|
|
+ struct message_header header;
|
|
|
+ int group_arut;
|
|
|
+ struct in_addr addr_arut;
|
|
|
+ short int fcc;
|
|
|
+ short int brake;
|
|
|
+ struct in_addr brake_addr;
|
|
|
+ struct rtr_item rtr_list[RTR_TOKEN_SIZE_MAX];
|
|
|
+ int rtr_list_entries;
|
|
|
+};
|
|
|
+
|
|
|
+struct conf_desc {
|
|
|
+ struct memb_conf_id conf_id;
|
|
|
+ int highest_seq;
|
|
|
+ int arut;
|
|
|
+#ifdef COMPLIE_OUT
|
|
|
+ int hole_list[HOLE_LIST_MAX];
|
|
|
+ int hole_list_entries;
|
|
|
+#endif
|
|
|
+};
|
|
|
+
|
|
|
+struct memb_form_token {
|
|
|
+ struct message_header header;
|
|
|
+ struct memb_conf_id conf_id;
|
|
|
+ struct conf_desc conf_desc_list[MAX_MEMBERS]; /* SHOULD BE MAX_MEMBERS */
|
|
|
+ int conf_desc_list_entries;
|
|
|
+ struct in_addr member_list[MAX_MEMBERS];
|
|
|
+ int member_list_entries;
|
|
|
+ struct in_addr rep_list[MAX_MEMBERS];
|
|
|
+ int rep_list_entries;
|
|
|
+};
|
|
|
+
|
|
|
+
|
|
|
+struct memb_attempt_join {
|
|
|
+ struct message_header header;
|
|
|
+};
|
|
|
+
|
|
|
+struct memb_join {
|
|
|
+ struct message_header header;
|
|
|
+ struct in_addr active_rep_list[MAX_MEMBERS];
|
|
|
+ int active_rep_list_entries;
|
|
|
+ struct in_addr failed_rep_list[MAX_MEMBERS];
|
|
|
+ int failed_rep_list_entries;
|
|
|
+};
|
|
|
+
|
|
|
+struct gmi_pend_trans_item {
|
|
|
+ struct mcast *mcast;
|
|
|
+
|
|
|
+ struct iovec iovec[MAXIOVS];
|
|
|
+ int iov_len;
|
|
|
+};
|
|
|
+
|
|
|
+struct gmi_pend_delv_item {
|
|
|
+ struct iovec iovec[MAXIOVS];
|
|
|
+ int iov_len;
|
|
|
+};
|
|
|
+
|
|
|
+struct gmi_rtr_item {
|
|
|
+ struct iovec iovec[MAXIOVS+2]; /* +2 is for mcast msg + group name TODO is this right */
|
|
|
+ int iov_len;
|
|
|
+};
|
|
|
+
|
|
|
+enum memb_state {
|
|
|
+ MEMB_STATE_OPERATIONAL,
|
|
|
+ MEMB_STATE_GATHER,
|
|
|
+ MEMB_STATE_COMMIT,
|
|
|
+ MEMB_STATE_FORM,
|
|
|
+ MEMB_STATE_EVS
|
|
|
+};
|
|
|
+
|
|
|
+static enum memb_state memb_state = MEMB_STATE_GATHER;
|
|
|
+
|
|
|
+static struct sockaddr_in memb_list[MAX_MEMBERS];
|
|
|
+static int memb_list_entries = 1;
|
|
|
+static int memb_list_entries_confchg = 1;
|
|
|
+
|
|
|
+struct sockaddr_in memb_next;
|
|
|
+
|
|
|
+struct in_addr memb_gather_set[MAX_MEMBERS];
|
|
|
+int memb_gather_set_entries = 0;
|
|
|
+
|
|
|
+struct memb_commit_set {
|
|
|
+ struct sockaddr_in rep;
|
|
|
+ struct in_addr join_rep_list[MAX_MEMBERS];
|
|
|
+ int join_rep_list_entries;
|
|
|
+ struct in_addr member_list[MAX_MEMBERS];
|
|
|
+ int member_list_entries;
|
|
|
+};
|
|
|
+
|
|
|
+static struct memb_commit_set memb_commit_set[MAX_MEMBERS];
|
|
|
+
|
|
|
+static int memb_commit_set_entries = 0;
|
|
|
+
|
|
|
+static struct in_addr memb_failed_list[MAX_MEMBERS];
|
|
|
+
|
|
|
+static int memb_failed_list_entries = 0;
|
|
|
+
|
|
|
+static struct sockaddr_in memb_local_sockaddr_in;
|
|
|
+
|
|
|
+static struct memb_conf_id memb_conf_id;
|
|
|
+
|
|
|
+static struct memb_conf_id memb_form_token_conf_id;
|
|
|
+
|
|
|
+static struct memb_join memb_join;
|
|
|
+
|
|
|
+static struct memb_form_token memb_form_token;
|
|
|
+
|
|
|
+char iov_buffer[MESSAGE_SIZE_MAX];
|
|
|
+
|
|
|
+static struct iovec gmi_iov_recv = {
|
|
|
+ iov_base: iov_buffer,
|
|
|
+ iov_len: sizeof (iov_buffer)
|
|
|
+};
|
|
|
+
|
|
|
+struct message_handlers {
|
|
|
+ int count;
|
|
|
+ int (*handler_functions[5]) (struct sockaddr_in *, struct iovec *, int, int);
|
|
|
+};
|
|
|
+
|
|
|
+poll_handle *gmi_poll_handle;
|
|
|
+
|
|
|
+void (*gmi_deliver_fn) (
|
|
|
+ struct gmi_groupname *groupname,
|
|
|
+ struct iovec *iovec,
|
|
|
+ int iov_len) = 0;
|
|
|
+
|
|
|
+void (*gmi_confchg_fn) (
|
|
|
+ struct sockaddr_in *member_list, int member_list_entries,
|
|
|
+ struct sockaddr_in *left_list, int left_list_entries,
|
|
|
+ struct sockaddr_in *joined_list, int joined_list_entries) = 0;
|
|
|
+
|
|
|
+/*
|
|
|
+ * forward decls
|
|
|
+ */
|
|
|
+static int message_handler_orf_token (struct sockaddr_in *, struct iovec *, int, int);
|
|
|
+static int message_handler_mcast (struct sockaddr_in *, struct iovec *, int, int);
|
|
|
+static int message_handler_memb_attempt_join (struct sockaddr_in *, struct iovec *, int, int);
|
|
|
+static int message_handler_memb_join (struct sockaddr_in *, struct iovec *, int, int);
|
|
|
+static int message_handler_memb_form_token (struct sockaddr_in *, struct iovec *, int, int);
|
|
|
+static void memb_conf_id_build (struct memb_conf_id *, struct in_addr);
|
|
|
+static int recv_handler (poll_handle handle, int fd, int revents, void *data);
|
|
|
+static int local_netif_determine (struct sockaddr_in *bindnet, struct sockaddr_in *bound_to);
|
|
|
+static int memb_state_gather_enter (void);
|
|
|
+static void pending_queues_deliver (void);
|
|
|
+static int orf_token_mcast (struct orf_token *orf_token,
|
|
|
+ int fcc_mcasts_allowed, struct sockaddr_in *system_from);
|
|
|
+static void queues_pend_delv_memb_new (void);
|
|
|
+static void calculate_group_arut (struct orf_token *orf_token);
|
|
|
+static int messages_free (int group_arut);
|
|
|
+
|
|
|
+struct message_handlers gmi_message_handlers = {
|
|
|
+ 5,
|
|
|
+ {
|
|
|
+ message_handler_orf_token,
|
|
|
+ message_handler_mcast,
|
|
|
+ message_handler_memb_attempt_join,
|
|
|
+ message_handler_memb_join,
|
|
|
+ message_handler_memb_form_token
|
|
|
+ }
|
|
|
+};
|
|
|
+
|
|
|
+/*
|
|
|
+ * Exported interfaces
|
|
|
+ */
|
|
|
+int gmi_init (
|
|
|
+ struct sockaddr_in *sockaddr_mcast,
|
|
|
+ struct sockaddr_in *sockaddr_bindnet,
|
|
|
+ poll_handle *poll_handle,
|
|
|
+ struct sockaddr_in *bound_to)
|
|
|
+{
|
|
|
+ int res;
|
|
|
+ struct ip_mreqn mreqn;
|
|
|
+ struct sockaddr_in sockaddr_in;
|
|
|
+ char flag;
|
|
|
+ int i;
|
|
|
+ int index;
|
|
|
+
|
|
|
+ memcpy (&sockaddr_in_mcast, sockaddr_mcast, sizeof (struct sockaddr_in));
|
|
|
+
|
|
|
+ for (i = 0; i < PRIORITY_MAX; i++) {
|
|
|
+ queue_init (&queues_pend_trans[i], QUEUE_PEND_TRANS_SIZE_MAX,
|
|
|
+ sizeof (struct gmi_pend_trans_item));
|
|
|
+ }
|
|
|
+ sq_init (&queue_rtr_items, QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct gmi_rtr_item), 0);
|
|
|
+ index = local_netif_determine (sockaddr_bindnet, bound_to);
|
|
|
+ if (index == -1) {
|
|
|
+ return (-1);
|
|
|
+ }
|
|
|
+ memcpy (&memb_list[0], &memb_local_sockaddr_in, sizeof (struct sockaddr_in));
|
|
|
+ memb_conf_id_build (&memb_conf_id, memb_local_sockaddr_in.sin_addr);
|
|
|
+ memcpy (&memb_form_token_conf_id, &memb_conf_id, sizeof (struct memb_conf_id));
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Set local sock addr for new socket
|
|
|
+ */
|
|
|
+ sockaddr_in.sin_family = AF_INET;
|
|
|
+ sockaddr_in.sin_addr.s_addr = htonl (INADDR_ANY);
|
|
|
+ sockaddr_in.sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Join group membership on socket
|
|
|
+ */
|
|
|
+ mreqn.imr_multiaddr.s_addr = sockaddr_mcast->sin_addr.s_addr;
|
|
|
+ mreqn.imr_ifindex = index + 1;
|
|
|
+
|
|
|
+ gmi_fd = socket (AF_INET, SOCK_DGRAM, 0);
|
|
|
+ if (gmi_fd == -1) {
|
|
|
+ perror ("socket");
|
|
|
+ return (-1);
|
|
|
+ }
|
|
|
+
|
|
|
+ res = bind (gmi_fd, (struct sockaddr *)&sockaddr_in,
|
|
|
+ sizeof (struct sockaddr_in));
|
|
|
+ if (res == -1) {
|
|
|
+ perror ("bind failed");
|
|
|
+ return (-1);
|
|
|
+ }
|
|
|
+#ifdef BROADCAST_CODE_TAKEN_OUT
|
|
|
+this code is the broadcast socket option vs multicast socket option
|
|
|
+ setsockopt (gmi_fd, SOL_SOCKET, SO_BROADCAST, (char *)&on, sizeof (on));
|
|
|
+#endif
|
|
|
+ res = setsockopt (gmi_fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
|
|
|
+ &mreqn, sizeof (mreqn));
|
|
|
+ if (res == -1) {
|
|
|
+ perror ("join multicast group failed");
|
|
|
+ return (-1);
|
|
|
+ }
|
|
|
+
|
|
|
+ flag = 0;
|
|
|
+ res = setsockopt (gmi_fd, IPPROTO_IP, IP_MULTICAST_LOOP,
|
|
|
+ &flag, sizeof (flag));
|
|
|
+ if (res == -1) {
|
|
|
+ perror ("turn off loopback");
|
|
|
+ return (-1);
|
|
|
+ }
|
|
|
+
|
|
|
+ gmi_poll_handle = poll_handle;
|
|
|
+
|
|
|
+ poll_dispatch_add (*gmi_poll_handle, gmi_fd, POLLIN, 0, recv_handler);
|
|
|
+
|
|
|
+ memb_state_gather_enter ();
|
|
|
+
|
|
|
+ memset (&memb_next, 0, sizeof (struct sockaddr_in));
|
|
|
+
|
|
|
+ queues_pend_delv_memb_new ();
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+int gmi_join (
|
|
|
+ struct gmi_groupname *groupname,
|
|
|
+ void (*deliver_fn) (
|
|
|
+ struct gmi_groupname *groupname,
|
|
|
+ struct iovec *iovec,
|
|
|
+ int iov_len),
|
|
|
+ void (*confchg_fn) (
|
|
|
+ struct sockaddr_in *member_list, int member_list_entries,
|
|
|
+ struct sockaddr_in *left_list, int left_list_entries,
|
|
|
+ struct sockaddr_in *joined_list, int joined_list_entries),
|
|
|
+ gmi_join_handle *handle_out) {
|
|
|
+
|
|
|
+ gmi_deliver_fn = deliver_fn;
|
|
|
+ gmi_confchg_fn = confchg_fn;
|
|
|
+ *handle_out = 0;
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+int local_host_seq_count = 0;
|
|
|
+
|
|
|
+int gmi_leave (
|
|
|
+ gmi_join_handle handle_join);
|
|
|
+
|
|
|
+static int gmi_pend_trans_item_store (
|
|
|
+ struct gmi_groupname *groupname,
|
|
|
+ struct iovec *iovec,
|
|
|
+ int iov_len,
|
|
|
+ int priority,
|
|
|
+ short packet_number, short packet_count)
|
|
|
+{
|
|
|
+ int i, j;
|
|
|
+ struct gmi_pend_trans_item gmi_pend_trans_item;
|
|
|
+ /*
|
|
|
+ * Store pending item
|
|
|
+ */
|
|
|
+ gmi_pend_trans_item.mcast = malloc (sizeof (struct mcast));
|
|
|
+ if (gmi_pend_trans_item.mcast == 0) {
|
|
|
+ goto error_mcast;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Set mcast header
|
|
|
+ */
|
|
|
+ gmi_pend_trans_item.mcast->header.type = MESSAGE_TYPE_MCAST;
|
|
|
+ gmi_pend_trans_item.mcast->priority = priority;
|
|
|
+ gmi_pend_trans_item.mcast->packet_number = packet_number;
|
|
|
+ gmi_pend_trans_item.mcast->packet_count = packet_count;
|
|
|
+ gmi_pend_trans_item.mcast->packet_seq = local_host_seq_count++;
|
|
|
+ gmi_pend_trans_item.mcast->source.s_addr = this_ip.sin_addr.s_addr;
|
|
|
+
|
|
|
+ memcpy (&gmi_pend_trans_item.mcast->groupname, groupname,
|
|
|
+ sizeof (struct gmi_groupname));
|
|
|
+
|
|
|
+ for (i = 0; i < iov_len; i++) {
|
|
|
+ gmi_pend_trans_item.iovec[i].iov_base = malloc (iovec[i].iov_len);
|
|
|
+ if (gmi_pend_trans_item.iovec[i].iov_base == 0) {
|
|
|
+ goto error_iovec;
|
|
|
+ }
|
|
|
+ memcpy (gmi_pend_trans_item.iovec[i].iov_base, iovec[i].iov_base,
|
|
|
+ iovec[i].iov_len);
|
|
|
+ gmi_pend_trans_item.iovec[i].iov_len = iovec[i].iov_len;
|
|
|
+ }
|
|
|
+ gmi_pend_trans_item.iov_len = iov_len;
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "mcasted message added to pending queue\n");
|
|
|
+ queue_item_add (&queues_pend_trans[priority], &gmi_pend_trans_item);
|
|
|
+
|
|
|
+ return (0);
|
|
|
+error_iovec:
|
|
|
+ for (j = 0; j < i; j++) {
|
|
|
+ free (gmi_pend_trans_item.iovec[j].iov_base);
|
|
|
+ }
|
|
|
+ return (-1);
|
|
|
+error_mcast:
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * 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 the algorithm
|
|
|
+ * and reassembled at the receiver.
|
|
|
+ */
|
|
|
+#define FRAGMENT_SIZE (1500 - sizeof (struct mcast) - 20 - 8)
|
|
|
+
|
|
|
+static void timer_function_single_member (void *data);
|
|
|
+
|
|
|
+static void single_member_deliver (void)
|
|
|
+{
|
|
|
+ struct orf_token orf_token;
|
|
|
+
|
|
|
+ memset (&orf_token, 0, sizeof (struct orf_token));
|
|
|
+ orf_token.header.seqid = gmi_arut;
|
|
|
+ orf_token.header.type = MESSAGE_TYPE_ORF_TOKEN;
|
|
|
+ orf_token.group_arut = gmi_arut;
|
|
|
+ orf_token.rtr_list_entries = 0;
|
|
|
+ orf_token_mcast (&orf_token, 99, &memb_local_sockaddr_in);
|
|
|
+ calculate_group_arut (&orf_token);
|
|
|
+ messages_free (gmi_arut);
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_single_member);
|
|
|
+ timer_form_token_timeout = 0;
|
|
|
+ poll_timer_add (*gmi_poll_handle, 1, 0,
|
|
|
+ timer_function_single_member, &timer_single_member);
|
|
|
+}
|
|
|
+
|
|
|
+static void timer_function_single_member (void *data)
|
|
|
+{
|
|
|
+ single_member_deliver ();
|
|
|
+}
|
|
|
+
|
|
|
+int gmi_mcast (
|
|
|
+ struct gmi_groupname *groupname,
|
|
|
+ struct iovec *iovec,
|
|
|
+ int iov_len,
|
|
|
+ int priority)
|
|
|
+{
|
|
|
+ int res;
|
|
|
+ struct iovec copied_iovec;
|
|
|
+ struct iovec pending_iovecs[MAXIOVS];
|
|
|
+ int pending_iovec_entries = 0;
|
|
|
+ int iovec_entry = 0;
|
|
|
+ int total_size;
|
|
|
+ int packet_size;
|
|
|
+ int i;
|
|
|
+ int packet_number = 0;
|
|
|
+ int packet_count = 0;
|
|
|
+
|
|
|
+ packet_size = FRAGMENT_SIZE;
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "MCASTING MESSAGE\n");
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine size of total message
|
|
|
+ */
|
|
|
+ total_size = 0;
|
|
|
+ for (i = 0; i < iov_len; i++) {
|
|
|
+ total_size += iovec[i].iov_len;
|
|
|
+ assert (iovec[i].iov_len < MESSAGE_SIZE_MAX);
|
|
|
+ }
|
|
|
+
|
|
|
+ packet_count = (total_size / packet_size);
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "Message size is %d\n", total_size);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Break message up into individual packets and publish them
|
|
|
+ */
|
|
|
+ copied_iovec.iov_base = iovec[0].iov_base;
|
|
|
+ copied_iovec.iov_len = iovec[0].iov_len;
|
|
|
+ packet_size = 0;
|
|
|
+ pending_iovec_entries = 0;
|
|
|
+ iovec_entry = 0;
|
|
|
+ do {
|
|
|
+ if (copied_iovec.iov_len + packet_size > FRAGMENT_SIZE) {
|
|
|
+ pending_iovecs[pending_iovec_entries].iov_base = copied_iovec.iov_base;
|
|
|
+ pending_iovecs[pending_iovec_entries].iov_len = FRAGMENT_SIZE - packet_size;
|
|
|
+ copied_iovec.iov_base += FRAGMENT_SIZE - packet_size;
|
|
|
+ copied_iovec.iov_len -= FRAGMENT_SIZE - packet_size;
|
|
|
+ packet_size += pending_iovecs[pending_iovec_entries].iov_len;
|
|
|
+ } else {
|
|
|
+ pending_iovecs[pending_iovec_entries].iov_base = copied_iovec.iov_base;
|
|
|
+ pending_iovecs[pending_iovec_entries].iov_len = copied_iovec.iov_len;
|
|
|
+ packet_size += copied_iovec.iov_len;
|
|
|
+ iovec_entry += 1; /* this must be before copied_iovec */
|
|
|
+ copied_iovec.iov_base = iovec[iovec_entry].iov_base;
|
|
|
+ copied_iovec.iov_len = iovec[iovec_entry].iov_len;
|
|
|
+ }
|
|
|
+ pending_iovec_entries += 1;
|
|
|
+ if (packet_size >= FRAGMENT_SIZE || packet_size == total_size) {
|
|
|
+#ifdef DEBUG
|
|
|
+for (i = 0; i < pending_iovec_entries; i++) {
|
|
|
+ assert (pending_iovecs[i].iov_len < MESSAGE_SIZE_MAX);
|
|
|
+ assert (pending_iovecs[i].iov_len >= 0);
|
|
|
+ printf ("iovecs[%d] %x %d\n", i, pending_iovecs[i].iov_base, pending_iovecs[i].iov_len);
|
|
|
+calced_total += pending_iovecs[i].iov_len;
|
|
|
+}
|
|
|
+printf ("CALCULATED TOTAL is %d\n", calced_total);
|
|
|
+#endif
|
|
|
+ total_size -= packet_size;
|
|
|
+ assert (total_size >= 0);
|
|
|
+ res = gmi_pend_trans_item_store (groupname, pending_iovecs,
|
|
|
+ pending_iovec_entries, priority, packet_number, packet_count);
|
|
|
+ pending_iovec_entries = 0;
|
|
|
+ iovec_entry = 0;
|
|
|
+ packet_size = 0;
|
|
|
+ packet_number += 1;
|
|
|
+ }
|
|
|
+ } while (total_size > 0);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * The queued messages are sent in orf_token_mcast, not this function
|
|
|
+ * But if this processor is the only node, it must deliver the messages
|
|
|
+ * for self-delivery requirements because orf_token_mcast is only called
|
|
|
+ * on reception of a token
|
|
|
+ */
|
|
|
+ if (memb_list_entries == 1) {
|
|
|
+ single_member_deliver ();
|
|
|
+ }
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+static int local_netif_determine (struct sockaddr_in *bindnet, struct sockaddr_in *bound_to)
|
|
|
+{
|
|
|
+ struct sockaddr_in *sockaddr_in;
|
|
|
+ int id_fd;
|
|
|
+ struct ifconf ifc;
|
|
|
+ int numreqs = 0;
|
|
|
+ int res;
|
|
|
+ int i;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Generate list of local interfaces in ifc.ifc_req structure
|
|
|
+ */
|
|
|
+ id_fd = socket (AF_INET, SOCK_DGRAM, 0);
|
|
|
+ ifc.ifc_buf = 0;
|
|
|
+ do {
|
|
|
+ numreqs += 8;
|
|
|
+ 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 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;
|
|
|
+ if ((sockaddr_in->sin_addr.s_addr & 0x00ffffff) == (bindnet->sin_addr.s_addr & 0x00ffffff)) {
|
|
|
+ memb_local_sockaddr_in.sin_addr.s_addr = sockaddr_in->sin_addr.s_addr;
|
|
|
+ memb_local_sockaddr_in.sin_family = AF_INET;
|
|
|
+ memb_local_sockaddr_in.sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ memcpy (bound_to, &memb_local_sockaddr_in, sizeof (struct sockaddr_in));
|
|
|
+ res = i;
|
|
|
+ break; /* for */
|
|
|
+ }
|
|
|
+ }
|
|
|
+ free (ifc.ifc_buf);
|
|
|
+ close (id_fd);
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * 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 seqid) {
|
|
|
+ struct msghdr msg_mcast;
|
|
|
+ struct gmi_rtr_item *gmi_rtr_item;
|
|
|
+ int res;
|
|
|
+ struct mcast *mcast;
|
|
|
+
|
|
|
+//printf ("remulticasting %d\n", seqid);
|
|
|
+ /*
|
|
|
+ * Get RTR item at seqid, if not available, return
|
|
|
+ */
|
|
|
+ res = sq_item_get (&queue_rtr_items, seqid, (void **)&gmi_rtr_item);
|
|
|
+ if (res != 0) {
|
|
|
+ return -1;
|
|
|
+ }
|
|
|
+ mcast = gmi_rtr_item->iovec[0].iov_base;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Build multicast message
|
|
|
+ */
|
|
|
+ msg_mcast.msg_name = &sockaddr_in_mcast;
|
|
|
+ msg_mcast.msg_namelen = sizeof (struct sockaddr_in);
|
|
|
+ msg_mcast.msg_iov = gmi_rtr_item->iovec;
|
|
|
+ msg_mcast.msg_iovlen = gmi_rtr_item->iov_len;
|
|
|
+ msg_mcast.msg_control = 0;
|
|
|
+ msg_mcast.msg_controllen = 0;
|
|
|
+ msg_mcast.msg_flags = 0;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Multicast message
|
|
|
+ */
|
|
|
+ res = sendmsg (gmi_fd, &msg_mcast, MSG_NOSIGNAL | MSG_DONTWAIT);
|
|
|
+ if (res == -1) {
|
|
|
+printf ("error during remulticast %d %d %d\n", seqid, errno, gmi_rtr_item->iov_len);
|
|
|
+ return (-1);
|
|
|
+ }
|
|
|
+ stats_sent += res;
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+int last_group_arut = 0;
|
|
|
+int last_released = 0;
|
|
|
+int set_arut = -1;
|
|
|
+
|
|
|
+/*
|
|
|
+ * Brake output multicasts if the missing window is too large
|
|
|
+ */
|
|
|
+int gmi_brake;
|
|
|
+
|
|
|
+static int messages_free (int group_arut)
|
|
|
+{
|
|
|
+ struct gmi_rtr_item *gmi_rtr_item_p;
|
|
|
+ int i, j;
|
|
|
+ int res;
|
|
|
+ int lesser;
|
|
|
+
|
|
|
+// TODO printf ("group arut %d last_group-arut %d gmi_dut %d barrier %d\n", group_arut, last_group_arut, gmi_dut, gmi_barrier_seq);
|
|
|
+ /*
|
|
|
+ * Determine braking value (when messages + MISSING_MCAST_WINDOW, stop sending messages)
|
|
|
+ */
|
|
|
+ gmi_brake = group_arut;
|
|
|
+ if (gmi_brake > last_group_arut) {
|
|
|
+ gmi_brake = last_group_arut;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine low water mark for messages to be freed
|
|
|
+ */
|
|
|
+ lesser = gmi_brake;
|
|
|
+ if (lesser > gmi_adut) {
|
|
|
+ lesser = gmi_adut;
|
|
|
+ }
|
|
|
+
|
|
|
+//printf ("Freeing lesser %d %d %d\n", lesser, group_arut, last_group_arut);
|
|
|
+//printf ("lesser %d gropu arut %d last group arut %d\n", lesser, group_arut, last_group_arut);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * return early if no messages can be freed
|
|
|
+ */
|
|
|
+/*
|
|
|
+ if (last_released + 1 == lesser) {
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+*/
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Release retransmit list items if group arut indicates they are transmitted
|
|
|
+ */
|
|
|
+ for (i = last_released; i <= lesser; i++) {
|
|
|
+ res = sq_item_get (&queue_rtr_items, i, (void **)&gmi_rtr_item_p);
|
|
|
+ if (res == 0) {
|
|
|
+ for (j = 0; j < gmi_rtr_item_p->iov_len; j++) {
|
|
|
+ free (gmi_rtr_item_p->iovec[j].iov_base);
|
|
|
+ gmi_rtr_item_p->iovec[j].iov_base = (void *)0xdeadbeef;
|
|
|
+ gmi_rtr_item_p->iovec[j].iov_len = i;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ last_released = i + 1;
|
|
|
+ }
|
|
|
+
|
|
|
+ sq_items_release (&queue_rtr_items, lesser);
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "releasing messages up to and including %d\n", lesser);
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Multicasts pending messages onto the ring (requires orf_token possession)
|
|
|
+ */
|
|
|
+static int orf_token_mcast (
|
|
|
+ struct orf_token *orf_token,
|
|
|
+ int fcc_mcasts_allowed,
|
|
|
+ struct sockaddr_in *system_from)
|
|
|
+{
|
|
|
+ struct msghdr msg_mcast;
|
|
|
+ struct gmi_rtr_item gmi_rtr_item;
|
|
|
+ struct gmi_pend_trans_item *gmi_pend_trans_item = 0;
|
|
|
+ int res = 0;
|
|
|
+ int orf_token_seqid;
|
|
|
+ struct mcast *mcast;
|
|
|
+ int last_packet = 1;
|
|
|
+ struct queue *queue_pend_trans;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Disallow multicasts unless state is operational
|
|
|
+ */
|
|
|
+ if (memb_state != MEMB_STATE_OPERATIONAL) {
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If received a token with a higher sequence number,
|
|
|
+ * set highest seq so retransmits can happen at end of
|
|
|
+ * message stream
|
|
|
+ */
|
|
|
+ if (orf_token->header.seqid > gmi_highest_seq) {
|
|
|
+ gmi_highest_seq = orf_token->header.seqid;
|
|
|
+ }
|
|
|
+
|
|
|
+ orf_token_seqid = orf_token->header.seqid;
|
|
|
+
|
|
|
+ queue_pend_trans = &queues_pend_trans[gmi_pend_queue_priority];
|
|
|
+
|
|
|
+ for (fcc_mcast_current = 0; fcc_mcast_current < fcc_mcasts_allowed; fcc_mcast_current++) {
|
|
|
+ /*
|
|
|
+ * determine which pending queue to take message
|
|
|
+ * from if this is not a message fragment
|
|
|
+ */
|
|
|
+ if (gmi_fragment == 0) {
|
|
|
+ gmi_pend_queue_priority = 0;
|
|
|
+ do {
|
|
|
+ queue_pend_trans = &queues_pend_trans[gmi_pend_queue_priority];
|
|
|
+
|
|
|
+ if (queue_is_empty (queue_pend_trans)) {
|
|
|
+ gmi_pend_queue_priority++;
|
|
|
+ } else {
|
|
|
+ break; /* from do - found first queue with data */
|
|
|
+ }
|
|
|
+ } while (gmi_pend_queue_priority < PRIORITY_MAX);
|
|
|
+ }
|
|
|
+
|
|
|
+ if (gmi_pend_queue_priority == PRIORITY_MAX) {
|
|
|
+ break; /* all queues are empty, break from for */
|
|
|
+ }
|
|
|
+// printf ("selecting pending queue %d\n", gmi_pend_queue_priority);
|
|
|
+
|
|
|
+ gmi_pend_trans_item = (struct gmi_pend_trans_item *)queue_item_get (queue_pend_trans);
|
|
|
+ /* preincrement required by algo */
|
|
|
+ gmi_pend_trans_item->mcast->header.seqid = ++orf_token->header.seqid;
|
|
|
+// UNDO printf ("multicasting seqid %d\n", gmi_pend_trans_item->mcast->header.seqid);
|
|
|
+
|
|
|
+ last_packet = (gmi_pend_trans_item->mcast->packet_number ==
|
|
|
+ gmi_pend_trans_item->mcast->packet_count);
|
|
|
+//printf ("last packet is %d current mcast %d\n", last_packet, fcc_mcast_current);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Build IO vector
|
|
|
+ */
|
|
|
+ gmi_rtr_item.iovec[0].iov_base = gmi_pend_trans_item->mcast;
|
|
|
+ gmi_rtr_item.iovec[0].iov_len = sizeof (struct mcast);
|
|
|
+
|
|
|
+ mcast = gmi_rtr_item.iovec[0].iov_base;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Is this a fragment of a message
|
|
|
+ */
|
|
|
+ if (mcast->packet_number == mcast->packet_count) {
|
|
|
+ gmi_fragment = 0;
|
|
|
+ } else {
|
|
|
+ gmi_fragment = 1;
|
|
|
+ }
|
|
|
+
|
|
|
+ memcpy (&mcast->memb_conf_id, &memb_form_token_conf_id,
|
|
|
+ sizeof (struct memb_conf_id));
|
|
|
+
|
|
|
+ memcpy (&gmi_rtr_item.iovec[1], gmi_pend_trans_item->iovec,
|
|
|
+ gmi_pend_trans_item->iov_len * sizeof (struct iovec));
|
|
|
+
|
|
|
+ gmi_rtr_item.iov_len = gmi_pend_trans_item->iov_len + 1;
|
|
|
+
|
|
|
+ assert (gmi_rtr_item.iov_len < 16);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Add message to retransmit queue
|
|
|
+ */
|
|
|
+ sq_item_add (&queue_rtr_items,
|
|
|
+ &gmi_rtr_item, gmi_pend_trans_item->mcast->header.seqid);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Delete item from pending queue
|
|
|
+ */
|
|
|
+ queue_item_remove (queue_pend_trans);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Build multicast message
|
|
|
+ */
|
|
|
+ msg_mcast.msg_name = &sockaddr_in_mcast;
|
|
|
+ msg_mcast.msg_namelen = sizeof (struct sockaddr_in);
|
|
|
+ msg_mcast.msg_iov = gmi_rtr_item.iovec;
|
|
|
+ msg_mcast.msg_iovlen = gmi_rtr_item.iov_len;
|
|
|
+ msg_mcast.msg_control = 0;
|
|
|
+ msg_mcast.msg_controllen = 0;
|
|
|
+ msg_mcast.msg_flags = 0;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Multicast message
|
|
|
+ */
|
|
|
+ res = sendmsg (gmi_fd, &msg_mcast, MSG_NOSIGNAL | MSG_DONTWAIT);
|
|
|
+ /*
|
|
|
+ * An error here is recovered by the multicast algorithm
|
|
|
+ */
|
|
|
+
|
|
|
+// TODO stats_sent isn't right below
|
|
|
+ stats_sent += res;
|
|
|
+ }
|
|
|
+ assert (fcc_mcast_current < 100);
|
|
|
+#ifdef OUTA
|
|
|
+ if (fcc_mcast_current > fcc_mcasts_allowed) {
|
|
|
+ fcc_mcast_current = fcc_mcasts_allowed;
|
|
|
+ }
|
|
|
+#endif
|
|
|
+ /*
|
|
|
+ * If messages mcasted, deliver any new messages to pending queues
|
|
|
+ */
|
|
|
+ if (fcc_mcast_current) {
|
|
|
+ if (gmi_pend_trans_item->mcast->header.seqid > gmi_highest_seq) {
|
|
|
+ gmi_highest_seq = gmi_pend_trans_item->mcast->header.seqid;
|
|
|
+ }
|
|
|
+ pending_queues_deliver ();
|
|
|
+//printf ("orf Token seqid is %d group %d\n", orf_token_seqid, orf_token->group_arut);
|
|
|
+#ifdef COMPILE_OUT
|
|
|
+ if (orf_token_seqid == orf_token->group_arut) {
|
|
|
+//printf ("previous group arut #1 %d\n", orf_token->group_arut);
|
|
|
+ orf_token->group_arut = orf_token_seqid + fcc_mcast_current;
|
|
|
+ orf_token->addr_arut.s_addr = 0;
|
|
|
+}
|
|
|
+//printf ("reasing group arut to %d\n", orf_token->group_arut);
|
|
|
+#endif
|
|
|
+ }
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * 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 void orf_token_rtr (
|
|
|
+ struct orf_token *orf_token,
|
|
|
+ int *fcc_allowed)
|
|
|
+{
|
|
|
+ int res;
|
|
|
+ int i, j;
|
|
|
+ int found;
|
|
|
+
|
|
|
+#ifdef COMPLE_OUT
|
|
|
+printf ("Retransmit List %d\n", orf_token->rtr_list_entries);
|
|
|
+for (i = 0; i < orf_token->rtr_list_entries; i++) {
|
|
|
+ printf ("%d ", orf_token->rtr_list[i].seqid);
|
|
|
+}
|
|
|
+printf ("\n");
|
|
|
+#endif
|
|
|
+
|
|
|
+ /*
|
|
|
+ * 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;) {
|
|
|
+#ifdef COMPILE_OUT
|
|
|
+printf ("%d.%d.%d vs %d.%d.%d\n",
|
|
|
+ orf_token->rtr_list[i].conf_id.rep.s_addr,
|
|
|
+ orf_token->rtr_list[i].conf_id.tv.tv_sec,
|
|
|
+ orf_token->rtr_list[i].conf_id.tv.tv_usec,
|
|
|
+ memb_form_token_conf_id.rep.s_addr,
|
|
|
+ memb_form_token_conf_id.tv.tv_sec,
|
|
|
+ memb_form_token_conf_id.tv.tv_usec);
|
|
|
+#endif
|
|
|
+ /*
|
|
|
+ * If this retransmit request isn't from this configuration,
|
|
|
+ * try next rtr entry
|
|
|
+ */
|
|
|
+ if (memcmp (&orf_token->rtr_list[i].conf_id, &memb_form_token_conf_id,
|
|
|
+ sizeof (struct memb_conf_id)) != 0) {
|
|
|
+
|
|
|
+ i++;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ assert (orf_token->rtr_list[i].seqid > 0);
|
|
|
+ res = orf_token_remcast (orf_token->rtr_list[i].seqid);
|
|
|
+ if (res == 0) {
|
|
|
+ orf_token->rtr_list_entries -= 1;
|
|
|
+ assert (orf_token->rtr_list_entries >= 0);
|
|
|
+ memmove (&orf_token->rtr_list[i],
|
|
|
+ &orf_token->rtr_list[i + 1],
|
|
|
+ sizeof (struct rtr_item) * (orf_token->rtr_list_entries));
|
|
|
+ fcc_remcast_current++;
|
|
|
+ stats_remcasts++;
|
|
|
+ } else {
|
|
|
+ i++;
|
|
|
+//printf ("couldn't remcast %d\n", i);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ *fcc_allowed = *fcc_allowed - fcc_remcast_current - 1;
|
|
|
+
|
|
|
+#ifdef COMPILE_OUT
|
|
|
+for (i = 0; i < orf_token->rtr_list_entries; i++) {
|
|
|
+ assert (orf_token->rtr_list[i].seqid != -1);
|
|
|
+}
|
|
|
+#endif
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Add messages to retransmit to RTR list
|
|
|
+ * but only retry if there is room in the retransmit list
|
|
|
+ */
|
|
|
+ for (i = gmi_arut + 1;
|
|
|
+ orf_token->rtr_list_entries < RTR_TOKEN_SIZE_MAX &&
|
|
|
+ // i <= orf_token->header.seqid; /* TODO this worked previously but not correct for EVS */
|
|
|
+ i <= gmi_highest_seq;
|
|
|
+ i++) {
|
|
|
+
|
|
|
+ res = sq_item_inuse (&queue_rtr_items, i);
|
|
|
+ if (res == 0) {
|
|
|
+ found = 0;
|
|
|
+ for (j = 0; j < orf_token->rtr_list_entries; j++) {
|
|
|
+ if (i == orf_token->rtr_list[j].seqid) {
|
|
|
+ found = 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (found == 0) {
|
|
|
+ memcpy (&orf_token->rtr_list[orf_token->rtr_list_entries].conf_id,
|
|
|
+ &memb_form_token_conf_id, sizeof (struct memb_conf_id));
|
|
|
+ orf_token->rtr_list[orf_token->rtr_list_entries].seqid = i;
|
|
|
+ orf_token->rtr_list_entries++;
|
|
|
+//printf ("adding to retransmit list %d\n", i);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Calculate flow control count
|
|
|
+ */
|
|
|
+static void orf_token_fcc (
|
|
|
+ struct orf_token *orf_token)
|
|
|
+{
|
|
|
+ orf_token->fcc = orf_token->fcc - fcc_mcast_last - fcc_remcast_last
|
|
|
+ + fcc_mcast_current + fcc_remcast_current;
|
|
|
+
|
|
|
+ fcc_mcast_last = fcc_mcast_current;
|
|
|
+ fcc_remcast_last = fcc_remcast_current;
|
|
|
+}
|
|
|
+
|
|
|
+static void queues_pend_delv_memb_new (void)
|
|
|
+{
|
|
|
+ struct pend_delv pend_delv_new[MAX_MEMBERS];
|
|
|
+ int item_index = 0;
|
|
|
+ int i, j;
|
|
|
+ int found;
|
|
|
+
|
|
|
+ memset (pend_delv_new, 0, sizeof (struct pend_delv) * MAX_MEMBERS);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Build new pending list
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ found = 0;
|
|
|
+ for (j = 0; j < MAX_MEMBERS; j++) {
|
|
|
+ /*
|
|
|
+ * If membership item in queues pending delivery list, copy it
|
|
|
+ */
|
|
|
+ if (memb_list[i].sin_addr.s_addr == queues_pend_delv[j].ip.s_addr) {
|
|
|
+ memcpy (&pend_delv_new[item_index], &queues_pend_delv[j],
|
|
|
+ sizeof (struct pend_delv));
|
|
|
+ item_index += 1;
|
|
|
+ found = 1;
|
|
|
+ break; /* for j = */
|
|
|
+ }
|
|
|
+ }
|
|
|
+ /*
|
|
|
+ * If membership item not found in pending delivery list, make new entry
|
|
|
+ */
|
|
|
+ if (found == 0) {
|
|
|
+ queue_init (&pend_delv_new[item_index].queue, QUEUE_PEND_DELV_SIZE_MAX,
|
|
|
+ sizeof (struct gmi_pend_delv_item));
|
|
|
+ pend_delv_new[item_index].seqid = 0;
|
|
|
+ pend_delv_new[item_index].ip.s_addr = memb_list[i].sin_addr.s_addr;
|
|
|
+ item_index += 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Copy new list into system list
|
|
|
+ */
|
|
|
+ memcpy (queues_pend_delv, pend_delv_new,
|
|
|
+ sizeof (struct pend_delv) * MAX_MEMBERS);
|
|
|
+
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ /*
|
|
|
+ * If queue not empty, mark it for first delivery
|
|
|
+ * otherwise reset seqno
|
|
|
+ */
|
|
|
+ if (queue_is_empty (&queues_pend_delv[i].queue) == 0) {
|
|
|
+ queues_pend_delv[i].first_delivery = 1;
|
|
|
+ } else {
|
|
|
+ queues_pend_delv[i].seqid = 0;
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+static int orf_token_evs (
|
|
|
+ struct orf_token *orf_token,
|
|
|
+ int starting_group_arut)
|
|
|
+{
|
|
|
+ int i, j;
|
|
|
+ struct sockaddr_in trans_memb_list[MAX_MEMBERS];
|
|
|
+ struct sockaddr_in left_list[MAX_MEMBERS];
|
|
|
+ struct sockaddr_in joined_list[MAX_MEMBERS];
|
|
|
+ int trans_memb_list_entries = 0;
|
|
|
+ int left_list_entries = 0;
|
|
|
+ int joined_list_entries = 0;
|
|
|
+ int found;
|
|
|
+
|
|
|
+//printf ("group arut is %d %d %d %d\n", orf_token->header.seqid, orf_token->group_arut, gmi_arut, gmi_highest_seq);
|
|
|
+ /*
|
|
|
+ * We should only execute this function if we are in EVS membership state
|
|
|
+ */
|
|
|
+ if (memb_state != MEMB_STATE_EVS) {
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Delete form token timer since the token has been swallowed
|
|
|
+ */
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_form_token_timeout);
|
|
|
+ timer_form_token_timeout = 0;
|
|
|
+
|
|
|
+printf ("EVS STATE group arut %d gmi arut %d highest %d barrier %d starting group arut %d\n", orf_token->group_arut, gmi_arut, gmi_highest_seq, gmi_barrier_seq, starting_group_arut);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * This node has reached highest seq, set local arut to barrier
|
|
|
+ */
|
|
|
+ if (gmi_arut == gmi_highest_seq) {
|
|
|
+//printf ("setting arut to barrier %d\n", gmi_barrier_seq);
|
|
|
+ gmi_arut = gmi_barrier_seq;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine when EVS recovery has completed
|
|
|
+ */
|
|
|
+//printf ("group arut is %d %d %d\n", orf_token->group_arut, gmi_arut, gmi_highest_seq);
|
|
|
+// TODO
|
|
|
+ if (memb_state == MEMB_STATE_EVS && gmi_arut == gmi_barrier_seq && orf_token->group_arut == gmi_barrier_seq) {
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "EVS recovery of messages complete, transitioning to operational.\n");
|
|
|
+ /*
|
|
|
+ * EVS recovery complete, reset local variables
|
|
|
+ */
|
|
|
+ gmi_arut = 0;
|
|
|
+
|
|
|
+ gmi_adut_old = gmi_adut;
|
|
|
+ gmi_adut = 0;
|
|
|
+
|
|
|
+ gmi_highest_seq_old = gmi_highest_seq;
|
|
|
+ gmi_highest_seq = 0;
|
|
|
+ last_group_arut = 0;
|
|
|
+ sq_reinit (&queue_rtr_items, 0);
|
|
|
+
|
|
|
+ memb_failed_list_entries = 0;
|
|
|
+
|
|
|
+ memb_state = MEMB_STATE_OPERATIONAL;
|
|
|
+ qsort (memb_form_token.member_list, memb_form_token.member_list_entries,
|
|
|
+ sizeof (struct in_addr), in_addr_compare);
|
|
|
+
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine transitional configuration
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ for (found = 0, j = 0; j < memb_form_token.member_list_entries; j++) {
|
|
|
+ if (memb_list[i].sin_addr.s_addr == memb_form_token.member_list[j].s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (found == 1) {
|
|
|
+ trans_memb_list[trans_memb_list_entries].sin_addr.s_addr = memb_list[i].sin_addr.s_addr;
|
|
|
+ trans_memb_list[trans_memb_list_entries].sin_family = AF_INET;
|
|
|
+ trans_memb_list[trans_memb_list_entries].sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ trans_memb_list_entries += 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine nodes that left the configuration
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ for (found = 0, j = 0; j < memb_form_token.member_list_entries; j++) {
|
|
|
+ if (memb_list[i].sin_addr.s_addr == memb_form_token.member_list[j].s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break; /* for j = 0 */
|
|
|
+ }
|
|
|
+ }
|
|
|
+ /*
|
|
|
+ * Node left membership, add it to list
|
|
|
+ */
|
|
|
+ if (found == 0) {
|
|
|
+ left_list[left_list_entries].sin_addr.s_addr = memb_list[i].sin_addr.s_addr;
|
|
|
+ left_list[left_list_entries].sin_family = AF_INET;
|
|
|
+ left_list[left_list_entries].sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ left_list_entries += 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * MAIN STEP:
|
|
|
+ * Deliver transitional configuration
|
|
|
+ */
|
|
|
+ if (gmi_confchg_fn &&
|
|
|
+ (trans_memb_list_entries != memb_list_entries ||
|
|
|
+ (memcmp (trans_memb_list, memb_list, sizeof (struct sockaddr_in) * memb_list_entries) != 0))) {
|
|
|
+ gmi_confchg_fn (trans_memb_list, trans_memb_list_entries,
|
|
|
+ left_list, left_list_entries,
|
|
|
+ 0, 0);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine nodes that joined the configuration
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_form_token.member_list_entries; i++) {
|
|
|
+ for (found = 0, j = 0; j < memb_list_entries_confchg; j++) {
|
|
|
+ if (memb_form_token.member_list[i].s_addr == memb_list[j].sin_addr.s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break; /* for j = 0 */
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Node joined membership, add it to list
|
|
|
+ */
|
|
|
+ if (found == 0) {
|
|
|
+ joined_list[joined_list_entries].sin_addr.s_addr = memb_form_token.member_list[i].s_addr;
|
|
|
+ joined_list[joined_list_entries].sin_family = AF_INET;
|
|
|
+ joined_list[joined_list_entries].sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ joined_list_entries += 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Install the form token's configuration into the local membership
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_form_token.member_list_entries; i++) {
|
|
|
+ memb_list[i].sin_addr.s_addr = memb_form_token.member_list[i].s_addr;
|
|
|
+ memb_list[i].sin_family = AF_INET;
|
|
|
+ memb_list[i].sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Install pending delivery queues
|
|
|
+ */
|
|
|
+ memb_list_entries = memb_form_token.member_list_entries;
|
|
|
+ memb_list_entries_confchg = memb_list_entries;
|
|
|
+ queues_pend_delv_memb_new ();
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Install new conf id
|
|
|
+ */
|
|
|
+ memcpy (&memb_conf_id, &memb_form_token.conf_id,
|
|
|
+ sizeof (struct memb_conf_id));
|
|
|
+ memcpy (&memb_form_token_conf_id, &memb_form_token.conf_id,
|
|
|
+ sizeof (struct memb_conf_id));
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Deliver regular configuration
|
|
|
+ */
|
|
|
+ if (gmi_confchg_fn) {
|
|
|
+ gmi_confchg_fn (memb_list, memb_list_entries,
|
|
|
+ left_list, 0,
|
|
|
+ joined_list, joined_list_entries);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+int gwin = 90;
|
|
|
+int pwin = 40;
|
|
|
+
|
|
|
+
|
|
|
+static int orf_fcc_allowed (struct orf_token *token)
|
|
|
+{
|
|
|
+ int allowed;
|
|
|
+
|
|
|
+ if (memb_state != MEMB_STATE_OPERATIONAL) {
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+ allowed = gwin + pwin - token->fcc;
|
|
|
+ if (allowed < 0) {
|
|
|
+ allowed = 0;
|
|
|
+ }
|
|
|
+ if (allowed > gwin) {
|
|
|
+ allowed = gwin;
|
|
|
+ }
|
|
|
+ if (allowed > pwin) {
|
|
|
+ allowed = pwin;
|
|
|
+ }
|
|
|
+ return (allowed);
|
|
|
+}
|
|
|
+
|
|
|
+void timer_function_form_token_timeout (void *data)
|
|
|
+{
|
|
|
+ log_printf (LOG_LEVEL_WARNING, "Token loss in FORM state\n");
|
|
|
+ memb_list_entries = 1;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Add highest rep to failed list to ensure termination
|
|
|
+ */
|
|
|
+ memb_failed_list[memb_failed_list_entries++].s_addr =
|
|
|
+ memb_form_token.rep_list[memb_form_token.rep_list_entries].s_addr;
|
|
|
+
|
|
|
+ memb_state_gather_enter ();
|
|
|
+}
|
|
|
+
|
|
|
+void orf_timer_function_token_timeout (void *data)
|
|
|
+{
|
|
|
+ switch (memb_state) {
|
|
|
+ case MEMB_STATE_OPERATIONAL:
|
|
|
+ log_printf (LOG_LEVEL_WARNING, "Token loss in OPERATIONAL.\n");
|
|
|
+ memb_conf_id.rep.s_addr = memb_local_sockaddr_in.sin_addr.s_addr;
|
|
|
+ memb_list_entries = 1;
|
|
|
+
|
|
|
+ memb_state_gather_enter ();
|
|
|
+ break;
|
|
|
+
|
|
|
+ case MEMB_STATE_GATHER:
|
|
|
+ case MEMB_STATE_COMMIT:
|
|
|
+ log_printf (LOG_LEVEL_WARNING, "Token loss in GATHER or COMMIT.\n");
|
|
|
+ memb_list_entries = 1;
|
|
|
+ break;
|
|
|
+
|
|
|
+ case MEMB_STATE_EVS:
|
|
|
+ log_printf (LOG_LEVEL_WARNING, "Token loss in EVS state\n");
|
|
|
+ memb_list_entries = 1;
|
|
|
+ memb_state_gather_enter ();
|
|
|
+ break;
|
|
|
+
|
|
|
+ default:
|
|
|
+ printf ("token loss in form state doesn't make sense here\n");
|
|
|
+ break;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Send orf_token to next member (requires orf_token)
|
|
|
+ */
|
|
|
+static int orf_token_send (
|
|
|
+ struct orf_token *orf_token)
|
|
|
+{
|
|
|
+ struct msghdr msg_orf_token;
|
|
|
+ struct iovec iovec_orf_token;
|
|
|
+ int res;
|
|
|
+
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_orf_token_timeout);
|
|
|
+
|
|
|
+ poll_timer_add (*gmi_poll_handle, TIMEOUT_TOKEN, 0,
|
|
|
+ orf_timer_function_token_timeout, &timer_orf_token_timeout);
|
|
|
+
|
|
|
+ iovec_orf_token.iov_base = orf_token;
|
|
|
+ iovec_orf_token.iov_len = sizeof (struct orf_token);
|
|
|
+
|
|
|
+ msg_orf_token.msg_name = &memb_next;
|
|
|
+ msg_orf_token.msg_namelen = sizeof (struct sockaddr_in);
|
|
|
+ msg_orf_token.msg_iov = &iovec_orf_token;
|
|
|
+ msg_orf_token.msg_iovlen = 1;
|
|
|
+ msg_orf_token.msg_control = 0;
|
|
|
+ msg_orf_token.msg_controllen = 0;
|
|
|
+ msg_orf_token.msg_flags = 0;
|
|
|
+
|
|
|
+// THIS IS FOR TESTING ERRORS IN THE EVS STATE
|
|
|
+//if ((memb_state == MEMB_STATE_EVS) && ((random () % 3) == 0)) {
|
|
|
+//log_printf (LOG_LEVEL_DEBUG, "CAUSING TOKEN LOSS AT EVS STATE\n");
|
|
|
+// return (1);
|
|
|
+//}
|
|
|
+
|
|
|
+ res = sendmsg (gmi_fd, &msg_orf_token, MSG_NOSIGNAL);
|
|
|
+ assert (res != -1);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * res not used here errors are handled by algorithm
|
|
|
+ */
|
|
|
+ gmi_last_seqid = orf_token->header.seqid;
|
|
|
+ stats_sent += res;
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+int orf_token_send_initial (void)
|
|
|
+{
|
|
|
+ struct orf_token orf_token;
|
|
|
+ int res;
|
|
|
+
|
|
|
+ orf_token.header.seqid = 0;
|
|
|
+ orf_token.header.type = MESSAGE_TYPE_ORF_TOKEN;
|
|
|
+ orf_token.group_arut = gmi_highest_seq;
|
|
|
+ orf_token.addr_arut.s_addr = this_ip.sin_addr.s_addr;
|
|
|
+ orf_token.fcc = 0;
|
|
|
+
|
|
|
+ orf_token.rtr_list_entries = 0;
|
|
|
+ memset (orf_token.rtr_list, 0, sizeof (struct rtr_item) * RTR_TOKEN_SIZE_MAX);
|
|
|
+
|
|
|
+ res = orf_token_send (&orf_token);
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Membership Management
|
|
|
+ */
|
|
|
+static int memb_join_send (void)
|
|
|
+{
|
|
|
+ struct msghdr msghdr_join;
|
|
|
+ struct iovec iovec_join;
|
|
|
+ int res;
|
|
|
+
|
|
|
+ memb_join.header.seqid = 0;
|
|
|
+ memb_join.header.type = MESSAGE_TYPE_MEMB_JOIN;
|
|
|
+ /*
|
|
|
+ * copy current gather list to representatives list
|
|
|
+ */
|
|
|
+ if ((memb_gather_set_entries == memb_join.active_rep_list_entries) &&
|
|
|
+ (memcmp (memb_join.active_rep_list, memb_gather_set,
|
|
|
+ sizeof (struct in_addr) * memb_gather_set_entries) == 0) &&
|
|
|
+ (memb_failed_list_entries == memb_join.failed_rep_list_entries) &&
|
|
|
+ (memcmp (memb_join.failed_rep_list, memb_failed_list,
|
|
|
+ sizeof (struct in_addr) * memb_failed_list_entries) == 0)) {
|
|
|
+
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Copy active reps
|
|
|
+ */
|
|
|
+ memcpy (memb_join.active_rep_list, memb_gather_set,
|
|
|
+ sizeof (struct in_addr) * memb_gather_set_entries);
|
|
|
+ memb_join.active_rep_list_entries = memb_gather_set_entries;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Copy failed reps
|
|
|
+ */
|
|
|
+ memcpy (memb_join.failed_rep_list, memb_failed_list,
|
|
|
+ sizeof (struct in_addr) * memb_failed_list_entries);
|
|
|
+ memb_join.failed_rep_list_entries = memb_failed_list_entries;
|
|
|
+
|
|
|
+ iovec_join.iov_base = &memb_join;
|
|
|
+ iovec_join.iov_len = sizeof (struct memb_join);
|
|
|
+
|
|
|
+ msghdr_join.msg_name = &sockaddr_in_mcast;
|
|
|
+ msghdr_join.msg_namelen = sizeof (struct sockaddr_in);
|
|
|
+ msghdr_join.msg_iov = &iovec_join;
|
|
|
+ msghdr_join.msg_iovlen = 1;
|
|
|
+ msghdr_join.msg_control = 0;
|
|
|
+ msghdr_join.msg_controllen = 0;
|
|
|
+ msghdr_join.msg_flags = 0;
|
|
|
+
|
|
|
+ res = sendmsg (gmi_fd, &msghdr_join, MSG_NOSIGNAL | MSG_DONTWAIT);
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+static int memb_state_commit_enter (void);
|
|
|
+
|
|
|
+/*
|
|
|
+ * Update gather_set[0].join_reps with list of failed members
|
|
|
+ */
|
|
|
+void memb_gather_set_update_failed (struct in_addr *list, int list_entries)
|
|
|
+{
|
|
|
+ int i;
|
|
|
+ int j;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Remove failed members from gather set
|
|
|
+ */
|
|
|
+ for (i = 0; i < list_entries; i++) {
|
|
|
+ for (j = 0; j < memb_gather_set_entries; j++) {
|
|
|
+ if (list[i].s_addr == memb_gather_set[j].s_addr) {
|
|
|
+ memb_gather_set_entries -= 1;
|
|
|
+ memcpy (&memb_gather_set[j],
|
|
|
+ &memb_gather_set[j + 1],
|
|
|
+ memb_gather_set_entries * sizeof (struct in_addr));
|
|
|
+ break; /* for j = 0 */
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+static void memb_timer_function_state_commit_timeout (void *data)
|
|
|
+{
|
|
|
+ int i;
|
|
|
+ int j;
|
|
|
+ int k;
|
|
|
+ int found;
|
|
|
+ int add_to_failed = 1;
|
|
|
+ struct sockaddr_in left_list[MAX_MEMBERS];
|
|
|
+ int left_list_entries = 0;
|
|
|
+
|
|
|
+ memb_failed_list_entries = 0;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * No entries responded in commit timeout period
|
|
|
+ */
|
|
|
+ if (memb_commit_set_entries == 0) {
|
|
|
+ /*
|
|
|
+ * memb_list_entries only set to 0 when token times out, in which case
|
|
|
+ * send a configuration change because no messages can be recovered in EVS
|
|
|
+ */
|
|
|
+ if (memb_list_entries == 0) {
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "I am the only member.\n");
|
|
|
+ if (gmi_confchg_fn) {
|
|
|
+ /*
|
|
|
+ * Determine nodes that left the configuration
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ if (memb_local_sockaddr_in.sin_addr.s_addr != memb_list[i].sin_addr.s_addr) {
|
|
|
+ left_list[left_list_entries].sin_addr.s_addr = memb_list[i].sin_addr.s_addr;
|
|
|
+ left_list[left_list_entries].sin_family = AF_INET;
|
|
|
+ left_list[left_list_entries].sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ left_list_entries += 1;
|
|
|
+
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ gmi_confchg_fn (&memb_local_sockaddr_in, 1,
|
|
|
+ left_list, left_list_entries,
|
|
|
+ 0, 0);
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "No members sent join, keeping old ring and transitioning to operational.\n");
|
|
|
+ }
|
|
|
+ memb_state = MEMB_STATE_OPERATIONAL;
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ /*
|
|
|
+ * Find all failed members
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_gather_set_entries; i++) {
|
|
|
+ add_to_failed = 1;
|
|
|
+ for (j = 0; j < memb_commit_set_entries; j++) {
|
|
|
+ /*
|
|
|
+ * If gather entry not in commit rep list, add to failed
|
|
|
+ */
|
|
|
+ if (memb_gather_set[i].s_addr == memb_commit_set[j].rep.sin_addr.s_addr) {
|
|
|
+ add_to_failed = 0;
|
|
|
+ break; /* for found = 0 */
|
|
|
+ }
|
|
|
+ }
|
|
|
+ /*
|
|
|
+ * If gather entry not in commit set, add to failed set
|
|
|
+ */
|
|
|
+ for (found = 0, j = 0; j < memb_commit_set_entries; j++) {
|
|
|
+ for (k = 0; k < memb_commit_set[j].join_rep_list_entries; k++) {
|
|
|
+ if (memb_gather_set[i].s_addr == memb_commit_set[j].join_rep_list[k].s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (found == 0) {
|
|
|
+ add_to_failed = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If local address, item found
|
|
|
+ */
|
|
|
+ if (memb_gather_set[i].s_addr == memb_local_sockaddr_in.sin_addr.s_addr) {
|
|
|
+ add_to_failed = 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (add_to_failed == 1) {
|
|
|
+ memb_failed_list[memb_failed_list_entries++].s_addr =
|
|
|
+ memb_gather_set[i].s_addr;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ memb_gather_set_update_failed (memb_failed_list, memb_failed_list_entries);
|
|
|
+
|
|
|
+ memb_state_commit_enter ();
|
|
|
+}
|
|
|
+
|
|
|
+static int memb_state_commit_enter (void)
|
|
|
+{
|
|
|
+ int res;
|
|
|
+
|
|
|
+ memb_state = MEMB_STATE_COMMIT;
|
|
|
+ memb_commit_set_entries = 0;
|
|
|
+ res = memb_join_send();
|
|
|
+
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_memb_state_gather_timeout);
|
|
|
+
|
|
|
+ timer_memb_state_gather_timeout = 0;
|
|
|
+
|
|
|
+ poll_timer_add (*gmi_poll_handle, TIMEOUT_STATE_COMMIT, 0,
|
|
|
+ memb_timer_function_state_commit_timeout, &timer_memb_state_commit_timeout);
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+static void memb_timer_function_state_gather (void *data)
|
|
|
+{
|
|
|
+ int i;
|
|
|
+ /*
|
|
|
+ * GATHER period expired, sort gather sets and send JOIN
|
|
|
+ */
|
|
|
+ memb_state_commit_enter ();
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "GATHER timeout:\n");
|
|
|
+ for (i = 0; i < memb_gather_set_entries; i++) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "host %d attempted to join %s\n", i, inet_ntoa (memb_gather_set[i]));
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+static void memb_print_commit_set (void)
|
|
|
+{
|
|
|
+ int i, j;
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "Gather list\n");
|
|
|
+ for (i = 0; i < memb_gather_set_entries; i++) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "\tmember %d %s\n", i, inet_ntoa (memb_gather_set[i]));
|
|
|
+ }
|
|
|
+ for (i = 0; i < memb_commit_set_entries; i++) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "Join from rep %d %s\n", i, inet_ntoa (memb_commit_set[i].rep.sin_addr));
|
|
|
+ for (j = 0; j < memb_commit_set[i].join_rep_list_entries; j++) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "\tmember %d %s\n", j, inet_ntoa (memb_commit_set[i].join_rep_list[j]));
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Determine if the commit phase has reached consensus
|
|
|
+ */
|
|
|
+static int memb_state_consensus_commit (void)
|
|
|
+{
|
|
|
+ int found;
|
|
|
+ int res;
|
|
|
+ int i, j;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine consensus
|
|
|
+ */
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If all commit sets don't match gather set, no consensus
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_commit_set_entries; i++) {
|
|
|
+ /*
|
|
|
+ * If not same number of entries, no consensus
|
|
|
+ */
|
|
|
+ res = memb_gather_set_entries - memb_commit_set[i].join_rep_list_entries;
|
|
|
+ if (res != 0) {
|
|
|
+ return (0); /* no consensus */
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If entries dont match, no consensus
|
|
|
+ */
|
|
|
+ res = memcmp (memb_gather_set, memb_commit_set[i].join_rep_list,
|
|
|
+ memb_gather_set_entries * sizeof (struct in_addr));
|
|
|
+
|
|
|
+ if (res != 0) {
|
|
|
+ return (0); /* no consensus */
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If all reps from gather set represented in commit set, consensus
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_gather_set_entries; i++) {
|
|
|
+ found = 0;
|
|
|
+ for (j = 0; j < memb_commit_set_entries; j++) {
|
|
|
+ if (memb_gather_set[i].s_addr == memb_local_sockaddr_in.sin_addr.s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ if (memb_gather_set[i].s_addr == memb_commit_set[j].rep.sin_addr.s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (found == 0) {
|
|
|
+ return (0); /* no consensus, rep not found from gather set */
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return (1); /* got consensus! */
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Union commit_set_entry into gather set
|
|
|
+ */
|
|
|
+static void memb_state_commit_union (int commit_set_entry)
|
|
|
+{
|
|
|
+int found;
|
|
|
+int i, j;
|
|
|
+
|
|
|
+ for (i = 0; i < memb_commit_set[commit_set_entry].join_rep_list_entries; i++) {
|
|
|
+ for (found = 0, j = 0; j < memb_gather_set_entries; j++) {
|
|
|
+ if (memb_commit_set[commit_set_entry].join_rep_list[i].s_addr ==
|
|
|
+ memb_gather_set[j].s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (found == 0) {
|
|
|
+ memb_gather_set[memb_gather_set_entries++].s_addr =
|
|
|
+ memb_commit_set[commit_set_entry].join_rep_list[i].s_addr;
|
|
|
+ /*
|
|
|
+ * Sort gather set
|
|
|
+ */
|
|
|
+ qsort (memb_gather_set, memb_gather_set_entries,
|
|
|
+ sizeof (struct in_addr), in_addr_compare);
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+static void memb_conf_id_build (
|
|
|
+ struct memb_conf_id *memb_conf_id,
|
|
|
+ struct in_addr memb_local_rep)
|
|
|
+{
|
|
|
+ gettimeofday (&memb_conf_id->tv, NULL);
|
|
|
+ memb_conf_id->rep.s_addr = memb_local_rep.s_addr;
|
|
|
+}
|
|
|
+
|
|
|
+static void memb_form_token_update_highest_seq (
|
|
|
+ struct memb_form_token *form_token)
|
|
|
+{
|
|
|
+ struct conf_desc *conf_desc;
|
|
|
+ int entry;
|
|
|
+ int found = 0;
|
|
|
+
|
|
|
+ for (entry = 0; entry < form_token->conf_desc_list_entries; entry++) {
|
|
|
+ if (memcmp (&form_token->conf_desc_list[entry].conf_id,
|
|
|
+ &memb_form_token_conf_id, sizeof (struct memb_conf_id)) == 0) {
|
|
|
+
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ conf_desc = &form_token->conf_desc_list[entry];
|
|
|
+ if (found && gmi_highest_seq < conf_desc->highest_seq) {
|
|
|
+ gmi_highest_seq = conf_desc->highest_seq;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+static void memb_form_token_conf_desc_build (
|
|
|
+ struct memb_form_token *form_token)
|
|
|
+{
|
|
|
+ struct conf_desc *conf_desc;
|
|
|
+ int found = 0;
|
|
|
+ int entry = 0;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine if local configuration id is already present in form token
|
|
|
+ */
|
|
|
+ for (entry = 0; entry < form_token->conf_desc_list_entries; entry++) {
|
|
|
+ if (memcmp (&form_token->conf_desc_list[entry].conf_id,
|
|
|
+ &memb_form_token_conf_id, sizeof (struct memb_conf_id)) == 0) {
|
|
|
+
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ conf_desc = &form_token->conf_desc_list[entry];
|
|
|
+
|
|
|
+ if (found == 0) {
|
|
|
+ /*
|
|
|
+ * Item not present, add item
|
|
|
+ */
|
|
|
+ conf_desc->highest_seq = gmi_highest_seq;
|
|
|
+ conf_desc->arut = gmi_arut;
|
|
|
+// TODO holes not currently implemented conf_desc->hole_list_entries = 0;
|
|
|
+ memcpy (&conf_desc->conf_id,
|
|
|
+ &memb_form_token_conf_id, sizeof (struct memb_conf_id));
|
|
|
+
|
|
|
+ form_token->conf_desc_list_entries += 1;
|
|
|
+ } else {
|
|
|
+ /*
|
|
|
+ * Item already present, update arut, highest seq
|
|
|
+ */
|
|
|
+ if (conf_desc->arut > gmi_arut) {
|
|
|
+ conf_desc->arut = gmi_arut;
|
|
|
+ }
|
|
|
+ if (gmi_highest_seq > conf_desc->highest_seq) {
|
|
|
+ conf_desc->highest_seq = gmi_highest_seq;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+#ifdef COMPILE_OUT
|
|
|
+ /*
|
|
|
+ * Build conf_desc->hole_list
|
|
|
+ */
|
|
|
+printf ("conf desc build %d %d\n", gmi_arut, gmi_highest_seq);
|
|
|
+ conf_desc->hole_list_entries = 0;
|
|
|
+ for (i = gmi_arut; i < gmi_highest_seq; i++) {
|
|
|
+ assert (conf_desc->hole_list_entries < HOLE_LIST_MAX);
|
|
|
+ res = sq_item_get (&queue_rtr_items, i, (void **)&gmi_rtr_item_p);
|
|
|
+ if (res == 0) {
|
|
|
+ /*
|
|
|
+ * If item present, delete from hole list if it exists
|
|
|
+ */
|
|
|
+ for (j = 0; j < conf_desc->hole_list_entries; j++) {
|
|
|
+ if (conf_desc->hole_list[j] == i) {
|
|
|
+ memmove (&conf_desc->hole_list[j], &conf_desc->hole_list[j + 1],
|
|
|
+ sizeof (int) * (conf_desc->hole_list_entries - j - 1));
|
|
|
+ conf_desc->hole_list_entries -= 1;
|
|
|
+printf ("reducing setting desc entries to %d\n", conf_desc->hole_list_entries);
|
|
|
+ break; /* from for (j = ... ) */
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ /*
|
|
|
+ * If item not present, add to hole list
|
|
|
+ */
|
|
|
+ conf_desc->hole_list[conf_desc->hole_list_entries] = i;
|
|
|
+ conf_desc->hole_list_entries += 1;
|
|
|
+printf ("increasing setting desc entries to %d %d\n", conf_desc->hole_list_entries, i);
|
|
|
+ }
|
|
|
+ }
|
|
|
+printf ("Conf desc build done\n");
|
|
|
+#endif
|
|
|
+}
|
|
|
+
|
|
|
+static int memb_form_token_send (
|
|
|
+ struct memb_form_token *form_token)
|
|
|
+{
|
|
|
+ struct msghdr msg_form_token;
|
|
|
+ struct iovec iovec_form_token;
|
|
|
+ int res;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Build message for sendmsg
|
|
|
+ */
|
|
|
+ iovec_form_token.iov_base = form_token;
|
|
|
+ iovec_form_token.iov_len = sizeof (struct memb_form_token);
|
|
|
+
|
|
|
+ msg_form_token.msg_name = &memb_next;
|
|
|
+ msg_form_token.msg_namelen = sizeof (struct sockaddr_in);
|
|
|
+ msg_form_token.msg_iov = &iovec_form_token;
|
|
|
+ msg_form_token.msg_iovlen = 1;
|
|
|
+ msg_form_token.msg_control = 0;
|
|
|
+ msg_form_token.msg_controllen = 0;
|
|
|
+ msg_form_token.msg_flags = 0;
|
|
|
+
|
|
|
+ res = sendmsg (gmi_fd, &msg_form_token, MSG_NOSIGNAL | MSG_DONTWAIT);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * res not used here, because orf token errors are handled by algorithm
|
|
|
+ */
|
|
|
+ stats_sent += res;
|
|
|
+
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_orf_token_timeout);
|
|
|
+ timer_orf_token_timeout = 0;
|
|
|
+
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_form_token_timeout);
|
|
|
+
|
|
|
+ poll_timer_add (*gmi_poll_handle, TIMEOUT_TOKEN, 0,
|
|
|
+ timer_function_form_token_timeout, &timer_form_token_timeout);
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+int memb_form_token_send_initial (void)
|
|
|
+{
|
|
|
+ struct memb_form_token form_token;
|
|
|
+ int res;
|
|
|
+
|
|
|
+ memset (&form_token, 0x00, sizeof (struct memb_form_token));
|
|
|
+ memb_state = MEMB_STATE_FORM;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Build form token
|
|
|
+ */
|
|
|
+ form_token.header.type = MESSAGE_TYPE_MEMB_FORM_TOKEN;
|
|
|
+ memcpy (form_token.rep_list,
|
|
|
+ memb_gather_set,
|
|
|
+ memb_gather_set_entries * sizeof (struct in_addr));
|
|
|
+ form_token.rep_list_entries = memb_gather_set_entries;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Add local member to entry
|
|
|
+ */
|
|
|
+ form_token.member_list[0].s_addr =
|
|
|
+ memb_local_sockaddr_in.sin_addr.s_addr;
|
|
|
+ form_token.member_list_entries = 1;
|
|
|
+
|
|
|
+ memb_conf_id_build (&form_token.conf_id, memb_local_sockaddr_in.sin_addr);
|
|
|
+
|
|
|
+ form_token.conf_desc_list_entries = 0;
|
|
|
+ memb_form_token_conf_desc_build (&form_token);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Send FORM to next member, or if no members in this configuration
|
|
|
+ * to next representative
|
|
|
+ */
|
|
|
+
|
|
|
+ if (memb_list_entries <= 1) {
|
|
|
+ memb_next.sin_addr.s_addr = memb_gather_set[1].s_addr;
|
|
|
+ }
|
|
|
+
|
|
|
+// TODO assertion here about the 1 value
|
|
|
+ memb_next.sin_family = AF_INET;
|
|
|
+ memb_next.sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+
|
|
|
+ res = memb_form_token_send (&form_token);
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+void print_stats (void)
|
|
|
+{
|
|
|
+ struct timeval tv_end;
|
|
|
+ gettimeofday (&tv_end, NULL);
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "Bytes recv %d\n", stats_recv);
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "Bytes sent %d\n", stats_sent);
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "Messages delivered %d\n", stats_delv);
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "Re-Mcasts %d\n", stats_remcasts);
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "Tokens process %d\n", stats_orf_token);
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Authenticates message using nonce, mac, and message body
|
|
|
+ */
|
|
|
+static int gmi_msg_auth (struct iovec *iovec, int iov_len)
|
|
|
+{
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+int last_lowered = 1;
|
|
|
+
|
|
|
+static void calculate_group_arut (struct orf_token *orf_token)
|
|
|
+{
|
|
|
+//printf ("group arut %d local arut %d gmi_gmi_highest seq %d\n", orf_token->group_arut, gmi_arut, gmi_highest_seq);
|
|
|
+//printf ("last %d group arut %d last arut %d arut %d\n", last_lowered, orf_token->group_arut, last_group_arut, gmi_arut);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * increase the group arut if we got back the same group
|
|
|
+ * because everyone has these messages
|
|
|
+ */
|
|
|
+ messages_free (orf_token->group_arut);
|
|
|
+ if (orf_token->addr_arut.s_addr == this_ip.sin_addr.s_addr) {
|
|
|
+ orf_token->group_arut = gmi_arut;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (gmi_arut < orf_token->group_arut) {
|
|
|
+ orf_token->group_arut = gmi_arut;
|
|
|
+ orf_token->addr_arut.s_addr = this_ip.sin_addr.s_addr;
|
|
|
+ }
|
|
|
+ last_group_arut = orf_token->group_arut;
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * Message Handlers
|
|
|
+ */
|
|
|
+
|
|
|
+/*
|
|
|
+ * 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)
|
|
|
+{
|
|
|
+ struct orf_token *orf_token;
|
|
|
+ int transmits_allowed;
|
|
|
+ int starting_group_arut;
|
|
|
+
|
|
|
+ orf_token = iovec[0].iov_base;
|
|
|
+
|
|
|
+#ifdef PRINT_STATS
|
|
|
+ if (orf_token->header.seqid > 10000) {
|
|
|
+ print_stats ();
|
|
|
+ }
|
|
|
+#endif
|
|
|
+
|
|
|
+ if (memb_state == MEMB_STATE_FORM) {
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "swallowing ORF token %d.\n", stats_orf_token);
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_orf_token_timeout);
|
|
|
+ timer_orf_token_timeout = 0;
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+//printf ("Got orf token from %s\n", inet_ntoa (system_from->sin_addr));
|
|
|
+ starting_group_arut = orf_token->group_arut;
|
|
|
+ stats_orf_token++;
|
|
|
+
|
|
|
+ transmits_allowed = orf_fcc_allowed (orf_token);
|
|
|
+
|
|
|
+//printf ("retransmit allowed %d\n", transmits_allowed);
|
|
|
+ /*
|
|
|
+ * Retransmit failed messages and request retransmissions
|
|
|
+ */
|
|
|
+
|
|
|
+ orf_token_rtr (orf_token, &transmits_allowed);
|
|
|
+//printf ("multicasts allowed %d\n", transmits_allowed);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * TODO Ok this is ugly and I dont like it.
|
|
|
+ *
|
|
|
+ * Flow control to limit number of missing multicast messages
|
|
|
+ * on lossy switches, this could cause a large window between
|
|
|
+ * what is delivered locally and what is delivered remotely.
|
|
|
+ * This window could cause the hole list of the form token to
|
|
|
+ * be overrun or cause the form token to be large.
|
|
|
+ */
|
|
|
+
|
|
|
+ if ((gmi_brake + MISSING_MCAST_WINDOW) < orf_token->header.seqid) {
|
|
|
+ transmits_allowed = 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Set the group arut and free any messages that can be freed
|
|
|
+ */
|
|
|
+ if (memb_state != MEMB_STATE_EVS) {
|
|
|
+ calculate_group_arut (orf_token);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Multicast queued messages
|
|
|
+ */
|
|
|
+ orf_token_mcast (orf_token, transmits_allowed, system_from);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Calculate flow control count
|
|
|
+ */
|
|
|
+ orf_token_fcc (orf_token);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Deliver membership and messages required by EVS
|
|
|
+ */
|
|
|
+ orf_token_evs (orf_token, starting_group_arut);
|
|
|
+
|
|
|
+ if (memb_state == MEMB_STATE_EVS) {
|
|
|
+ calculate_group_arut (orf_token);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Transmit orf_token to next member
|
|
|
+ */
|
|
|
+ orf_token_send (orf_token);
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+static int memb_state_gather_enter (void) {
|
|
|
+ struct msghdr msghdr_attempt_join;
|
|
|
+ struct iovec iovec_attempt_join;
|
|
|
+ struct memb_attempt_join memb_attempt_join;
|
|
|
+ int res = 0;
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "entering GATHER state.\n");
|
|
|
+ memb_state = MEMB_STATE_GATHER;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Join message starts with no entries
|
|
|
+ */
|
|
|
+ memb_join.active_rep_list_entries = 0;
|
|
|
+ memb_join.failed_rep_list_entries = 0;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Copy local host info
|
|
|
+ */
|
|
|
+ memb_gather_set[0].s_addr = memb_local_sockaddr_in.sin_addr.s_addr;
|
|
|
+ memb_gather_set_entries = 1;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If this node is the representative, send attempt join
|
|
|
+ */
|
|
|
+ if (memb_local_sockaddr_in.sin_addr.s_addr == memb_conf_id.rep.s_addr) {
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "SENDING attempt join because this node is ring rep.\n");
|
|
|
+ memb_attempt_join.header.seqid = 0;
|
|
|
+ memb_attempt_join.header.type = MESSAGE_TYPE_MEMB_ATTEMPT_JOIN;
|
|
|
+
|
|
|
+ iovec_attempt_join.iov_base = &memb_attempt_join;
|
|
|
+ iovec_attempt_join.iov_len = sizeof (struct memb_attempt_join);
|
|
|
+
|
|
|
+ msghdr_attempt_join.msg_name = &sockaddr_in_mcast;
|
|
|
+ msghdr_attempt_join.msg_namelen = sizeof (struct sockaddr_in);
|
|
|
+ msghdr_attempt_join.msg_iov = &iovec_attempt_join;
|
|
|
+ msghdr_attempt_join.msg_iovlen = 1;
|
|
|
+ msghdr_attempt_join.msg_control = 0;
|
|
|
+ msghdr_attempt_join.msg_controllen = 0;
|
|
|
+ msghdr_attempt_join.msg_flags = 0;
|
|
|
+
|
|
|
+ res = sendmsg (gmi_fd, &msghdr_attempt_join, MSG_NOSIGNAL | MSG_DONTWAIT);
|
|
|
+ /*
|
|
|
+ * res not checked here, there is nothing that can be done
|
|
|
+ * instead rely on the algorithm to recover from faults
|
|
|
+ */
|
|
|
+ }
|
|
|
+
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_memb_state_gather_timeout);
|
|
|
+
|
|
|
+ poll_timer_add (*gmi_poll_handle, TIMEOUT_STATE_GATHER, 0,
|
|
|
+ memb_timer_function_state_gather, &timer_memb_state_gather_timeout);
|
|
|
+
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+struct pend_delv *pend_delv_next_delivery_find (void)
|
|
|
+{
|
|
|
+ struct pend_delv *pend_delv = 0;
|
|
|
+ int i;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Find first_delivery queue that is not mepty
|
|
|
+ * this sets the first pend_delv
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ if (queues_pend_delv[i].first_delivery &&
|
|
|
+ queue_is_empty (&queues_pend_delv[i].queue) == 0) {
|
|
|
+
|
|
|
+ pend_delv = &queues_pend_delv[i];
|
|
|
+// printf ("Selecting first queue %s\n", inet_ntoa (pend_delv->ip));
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Search remaining pend_delv for first deliveries with
|
|
|
+ * smaller sequence numbers
|
|
|
+ */
|
|
|
+ for (++i; i < memb_list_entries_confchg; i++) {
|
|
|
+ assert (pend_delv);
|
|
|
+ if (queues_pend_delv[i].first_delivery &&
|
|
|
+ (queue_is_empty (&queues_pend_delv[i].queue) == 0) &&
|
|
|
+ (queues_pend_delv[i].seqid < pend_delv->seqid)) {
|
|
|
+
|
|
|
+ pend_delv = &queues_pend_delv[i];
|
|
|
+// printf ("Selecting first queue %s\n", inet_ntoa (pend_delv->ip));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Found first_delivery queue that wasn't empty, return it
|
|
|
+ */
|
|
|
+ if (pend_delv) {
|
|
|
+ return (pend_delv);
|
|
|
+ }
|
|
|
+ /*
|
|
|
+ * No first delivery queues, repeat same
|
|
|
+ * process looking for any queue
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ if (queue_is_empty (&queues_pend_delv[i].queue) == 0) {
|
|
|
+ pend_delv = &queues_pend_delv[i];
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Find lowest sequence number queue
|
|
|
+ */
|
|
|
+ for (++i; i < memb_list_entries_confchg; i++) {
|
|
|
+ assert (pend_delv);
|
|
|
+ if ((queue_is_empty (&queues_pend_delv[i].queue) == 0) &&
|
|
|
+ (queues_pend_delv[i].seqid < pend_delv->seqid)) {
|
|
|
+ pend_delv = &queues_pend_delv[i];
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return (pend_delv);
|
|
|
+}
|
|
|
+
|
|
|
+static int user_deliver ()
|
|
|
+{
|
|
|
+ struct gmi_pend_delv_item *pend_delv_item;
|
|
|
+ int i = 0;
|
|
|
+ int res = 0;
|
|
|
+ struct iovec iovec_delv[256];
|
|
|
+ int iov_len_delv = 0;
|
|
|
+ struct mcast *mcast = 0;
|
|
|
+ int messages_delivered = 0;
|
|
|
+ struct pend_delv *pend_delv;
|
|
|
+ int retval = 0;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Find pend_delv with lowest sequence number. This pend_delv is
|
|
|
+ * the queue that should be delivered from next
|
|
|
+ */
|
|
|
+ pend_delv = pend_delv_next_delivery_find ();
|
|
|
+//printf ("Delivering from queue %s\n", inet_ntoa (pend_delv->ip));
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If a message was not assembled on the queue with the lowest
|
|
|
+ * sequence number, return since there is no reason to attempt assembly.
|
|
|
+ */
|
|
|
+ memset (iovec_delv, 0, sizeof (iovec_delv));
|
|
|
+ queue_item_iterator_init (&pend_delv->queue);
|
|
|
+ assert (queue_is_empty (&pend_delv->queue) == 0);
|
|
|
+//printf ("Starting a packet assembly\n");
|
|
|
+ do {
|
|
|
+ pend_delv_item = queue_item_iterator_get (&pend_delv->queue);
|
|
|
+ mcast = pend_delv_item->iovec[0].iov_base;
|
|
|
+
|
|
|
+ assert (pend_delv_item);
|
|
|
+ assert (pend_delv_item->iovec[0].iov_len < MESSAGE_SIZE_MAX);
|
|
|
+ assert (pend_delv_item->iovec[0].iov_len != 0);
|
|
|
+ assert (pend_delv_item->iovec[0].iov_base != 0);
|
|
|
+ assert (mcast != (struct mcast *)0xdeadbeef);
|
|
|
+ assert (pend_delv->ip.s_addr == mcast->source.s_addr);
|
|
|
+
|
|
|
+ messages_delivered += 1;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Assemble io vector
|
|
|
+ */
|
|
|
+ if (pend_delv_item->iovec[0].iov_len == sizeof (struct mcast)) {
|
|
|
+ /*
|
|
|
+ * Copy iovec from second iovec if this is self-delivered
|
|
|
+ */
|
|
|
+ memcpy (&iovec_delv[iov_len_delv],
|
|
|
+ &pend_delv_item->iovec[1],
|
|
|
+ sizeof (struct iovec) * pend_delv_item->iov_len - 1);
|
|
|
+ iov_len_delv += pend_delv_item->iov_len - 1;
|
|
|
+ } else {
|
|
|
+ /*
|
|
|
+ * Copy iovec from first iovec if this is an external message
|
|
|
+ */
|
|
|
+ iovec_delv[iov_len_delv].iov_base =
|
|
|
+ pend_delv_item->iovec[0].iov_base + sizeof (struct mcast);
|
|
|
+ iovec_delv[iov_len_delv].iov_len =
|
|
|
+ pend_delv_item->iovec[0].iov_len - sizeof (struct mcast);
|
|
|
+ assert (iovec_delv[iov_len_delv].iov_len < MESSAGE_SIZE_MAX);
|
|
|
+ iov_len_delv += 1;
|
|
|
+ if (pend_delv_item->iov_len > 1) {
|
|
|
+ memcpy (&iovec_delv[iov_len_delv],
|
|
|
+ &pend_delv_item->iovec[1],
|
|
|
+ sizeof (struct iovec) * pend_delv_item->iov_len - 1);
|
|
|
+ iov_len_delv += pend_delv_item->iov_len - 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ assert (iov_len_delv < 256);
|
|
|
+ assert (iov_len_delv > 0);
|
|
|
+
|
|
|
+//printf ("Assembling from packet %d of %d of total %d\n",
|
|
|
+// mcast->packet_number, mcast->packet_count, mcast->packet_seq);
|
|
|
+ /*
|
|
|
+ * Deliver message if this is the last packet
|
|
|
+ */
|
|
|
+ if (mcast->packet_number == mcast->packet_count) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "Last packet, delivering iovec %d entries seq %d\n",
|
|
|
+ iov_len_delv, i);
|
|
|
+
|
|
|
+ gmi_deliver_fn (
|
|
|
+ &mcast->groupname,
|
|
|
+ iovec_delv,
|
|
|
+ iov_len_delv);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * On the first message delivery:
|
|
|
+ * Free items in the pending queue up to the barrier message
|
|
|
+ * set gmi_adut to rut so that message_free may free any messages.
|
|
|
+ */
|
|
|
+ if (pend_delv->first_delivery) {
|
|
|
+// printf ("releasing all messages up to %d\n", gmi_adut);
|
|
|
+// TODO actually release the messages from the previous configuration
|
|
|
+// TODO without a fix here, those messages are leaked
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Because of the ordering guarantees, we are guaranteed that
|
|
|
+ * pend_delv->seqid on every invocation of user_deliver shall
|
|
|
+ * increase (or reset to zero). This allows us to set the
|
|
|
+ * low water mark (gmi_adut) for freeing of messages to atleast
|
|
|
+ * the beginning of this message.
|
|
|
+ */
|
|
|
+ gmi_adut = pend_delv->seqid;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine if there are more messages on this queue
|
|
|
+ */
|
|
|
+ res = queue_item_iterator_next (&pend_delv->queue);
|
|
|
+ if (res == 0) {
|
|
|
+ /*
|
|
|
+ * More items to deliver set queues seqid head so
|
|
|
+ * correct pending queue can be selected next time
|
|
|
+ */
|
|
|
+ pend_delv_item = queue_item_iterator_get (&pend_delv->queue);
|
|
|
+ mcast = pend_delv_item->iovec[0].iov_base;
|
|
|
+ pend_delv->seqid = mcast->header.seqid;
|
|
|
+ for (i = 0; i < messages_delivered; i++) {
|
|
|
+ queue_item_remove (&pend_delv->queue);
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ /*
|
|
|
+ * No more items to deliver
|
|
|
+ */
|
|
|
+ pend_delv->seqid = 0;
|
|
|
+ queue_reinit (&pend_delv->queue);
|
|
|
+ }
|
|
|
+
|
|
|
+ retval = 1;
|
|
|
+ break; /* From do loop */
|
|
|
+ }
|
|
|
+
|
|
|
+ res = queue_item_iterator_next (&pend_delv->queue);
|
|
|
+ } while (res == 0);
|
|
|
+ return (retval);
|
|
|
+}
|
|
|
+
|
|
|
+struct pend_delv *pend_delv_find (struct in_addr source)
|
|
|
+{
|
|
|
+ struct pend_delv *pend_delv = 0;
|
|
|
+ int i;
|
|
|
+
|
|
|
+ for (i = 0; i < memb_list_entries_confchg; i++) {
|
|
|
+ if (source.s_addr == queues_pend_delv[i].ip.s_addr) {
|
|
|
+ pend_delv = &queues_pend_delv[i];
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return (pend_delv);
|
|
|
+}
|
|
|
+
|
|
|
+static int delivery_outstanding = 0;
|
|
|
+
|
|
|
+static void pending_queues_deliver (void)
|
|
|
+{
|
|
|
+ struct gmi_rtr_item *gmi_rtr_item_p;
|
|
|
+ int i;
|
|
|
+ int res;
|
|
|
+ struct mcast *mcast;
|
|
|
+ struct gmi_pend_delv_item pend_delv_item;
|
|
|
+ struct pend_delv *pend_delv;
|
|
|
+ int delivered;
|
|
|
+
|
|
|
+//printf ("Delivering messages to pending queues\n");
|
|
|
+ /*
|
|
|
+ * Deliver messages in order from rtr queue to pending delivery queue
|
|
|
+ */
|
|
|
+ for (i = gmi_arut + 1; i <= gmi_highest_seq; i++) {
|
|
|
+ res = sq_item_get (&queue_rtr_items, i, (void **)&gmi_rtr_item_p);
|
|
|
+ /*
|
|
|
+ * If hole, stop assembly
|
|
|
+ */
|
|
|
+ if (res != 0) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ assert (gmi_rtr_item_p->iovec[0].iov_len < MESSAGE_SIZE_MAX);
|
|
|
+ mcast = gmi_rtr_item_p->iovec[0].iov_base;
|
|
|
+ if (mcast == (struct mcast *)0xdeadbeef) {
|
|
|
+ printf ("seqid %d\n", gmi_rtr_item_p->iovec[0].iov_len);
|
|
|
+ }
|
|
|
+ assert (mcast != (struct mcast *)0xdeadbeef);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Message found
|
|
|
+ */
|
|
|
+ log_printf (LOG_LEVEL_DEBUG,
|
|
|
+ "Delivering MCAST message with seqid %d to pending delivery queue\n",
|
|
|
+ mcast->header.seqid);
|
|
|
+
|
|
|
+//printf ("Delivering MCAST from packet %d of %d of total %d seqid %d\n", mcast->packet_number, mcast->packet_count, mcast->packet_seq, mcast->header.seqid);
|
|
|
+ gmi_arut = i;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Create pending delivery item
|
|
|
+ */
|
|
|
+ pend_delv_item.iov_len = gmi_rtr_item_p->iov_len;
|
|
|
+ memcpy (&pend_delv_item.iovec, gmi_rtr_item_p->iovec,
|
|
|
+ sizeof (struct iovec) * gmi_rtr_item_p->iov_len);
|
|
|
+ assert (gmi_rtr_item_p->iov_len < MAXIOVS);
|
|
|
+
|
|
|
+ assert (mcast->source.s_addr != 0);
|
|
|
+ pend_delv = pend_delv_find (mcast->source);
|
|
|
+ assert (pend_delv != 0);
|
|
|
+ assert (pend_delv->ip.s_addr != 0);
|
|
|
+
|
|
|
+ if (mcast->packet_number == 0) {
|
|
|
+ pend_delv->seqid = mcast->header.seqid;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Add pending delivery item to pending delivery queue
|
|
|
+ */
|
|
|
+ queue_item_add (&pend_delv->queue, &pend_delv_item);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If message is complete, attempt delivery of all messages
|
|
|
+ * that are currently outstanding
|
|
|
+ */
|
|
|
+ if (mcast->packet_number == mcast->packet_count) {
|
|
|
+//printf ("Starting delivery\n");
|
|
|
+ delivery_outstanding += 1;
|
|
|
+ do {
|
|
|
+ delivered = user_deliver ();
|
|
|
+ if (delivered) {
|
|
|
+ delivery_outstanding -= 1;
|
|
|
+ }
|
|
|
+ } while (delivery_outstanding && delivered);
|
|
|
+ }
|
|
|
+ }
|
|
|
+//printf ("Done delivering messages to pending queues\n");
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * 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)
|
|
|
+{
|
|
|
+ struct gmi_rtr_item gmi_rtr_item;
|
|
|
+ struct mcast *mcast;
|
|
|
+
|
|
|
+ mcast = iovec[0].iov_base;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Ignore multicasts for other configurations
|
|
|
+ * TODO shouldn't we enter gather here?
|
|
|
+ */
|
|
|
+ if (memcmp (&mcast->memb_conf_id,
|
|
|
+ &memb_form_token_conf_id, sizeof (struct memb_conf_id)) != 0) {
|
|
|
+
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Add mcast message to rtr queue if not already in rtr queue
|
|
|
+ * otherwise free io vectors
|
|
|
+ */
|
|
|
+ if (bytes_received > 0 && bytes_received < MESSAGE_SIZE_MAX &&
|
|
|
+ sq_item_inuse (&queue_rtr_items, mcast->header.seqid) == 0) {
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Allocate new multicast memory block
|
|
|
+ * TODO we need to free this somewhere
|
|
|
+ */
|
|
|
+ gmi_rtr_item.iovec[0].iov_base = malloc (bytes_received);
|
|
|
+ if (gmi_rtr_item.iovec[0].iov_base == 0) {
|
|
|
+ return (-1); /* error here is corrected by the algorithm */
|
|
|
+ }
|
|
|
+ memcpy (gmi_rtr_item.iovec[0].iov_base, mcast, bytes_received);
|
|
|
+ gmi_rtr_item.iovec[0].iov_len = bytes_received;
|
|
|
+ assert (gmi_rtr_item.iovec[0].iov_len > 0);
|
|
|
+ assert (gmi_rtr_item.iovec[0].iov_len < MESSAGE_SIZE_MAX);
|
|
|
+ gmi_rtr_item.iov_len = 1;
|
|
|
+
|
|
|
+ if (mcast->header.seqid > gmi_highest_seq) {
|
|
|
+ gmi_highest_seq = mcast->header.seqid;
|
|
|
+ }
|
|
|
+
|
|
|
+ sq_item_add (&queue_rtr_items, &gmi_rtr_item, mcast->header.seqid);
|
|
|
+ }
|
|
|
+
|
|
|
+ pending_queues_deliver ();
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+static int message_handler_memb_attempt_join (
|
|
|
+ struct sockaddr_in *system_from,
|
|
|
+ struct iovec *iov,
|
|
|
+ int iov_len,
|
|
|
+ int bytes_received)
|
|
|
+{
|
|
|
+ int token_lost;
|
|
|
+ int found;
|
|
|
+ int i;
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "Got attempt join from %s\n", inet_ntoa (system_from->sin_addr));
|
|
|
+
|
|
|
+ for (token_lost = 0, i = 0; i < memb_list_entries; i++) {
|
|
|
+ if (memb_list[i].sin_addr.s_addr == system_from->sin_addr.s_addr &&
|
|
|
+ memb_conf_id.rep.s_addr != system_from->sin_addr.s_addr) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "ATTEMPT JOIN, token lost, taking attempt join msg.\n");
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_orf_token_timeout);
|
|
|
+ timer_orf_token_timeout = 0;
|
|
|
+ memb_conf_id.rep.s_addr = memb_local_sockaddr_in.sin_addr.s_addr;
|
|
|
+ token_lost = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Not representative
|
|
|
+ */
|
|
|
+ if (token_lost == 0 &&
|
|
|
+ memb_conf_id.rep.s_addr != memb_local_sockaddr_in.sin_addr.s_addr) {
|
|
|
+
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "not the rep for this ring, not handling attempt join.\n");
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+ switch (memb_state) {
|
|
|
+ case MEMB_STATE_OPERATIONAL:
|
|
|
+ case MEMB_STATE_COMMIT:
|
|
|
+ memb_state_gather_enter ();
|
|
|
+ /*
|
|
|
+ * Do NOT place break here, immediately execute gather attempt join
|
|
|
+ */
|
|
|
+
|
|
|
+ case MEMB_STATE_GATHER:
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "ATTEMPT JOIN: state gather\n");
|
|
|
+ for (found = 0, i = 0; i < memb_gather_set_entries; i++) {
|
|
|
+ if (memb_gather_set[i].s_addr == system_from->sin_addr.s_addr) {
|
|
|
+ found = 1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (found == 0) {
|
|
|
+ memb_gather_set[memb_gather_set_entries++].s_addr = system_from->sin_addr.s_addr;
|
|
|
+ /*
|
|
|
+ * Sort gather set
|
|
|
+ */
|
|
|
+ qsort (memb_gather_set, memb_gather_set_entries,
|
|
|
+ sizeof (struct in_addr), in_addr_compare);
|
|
|
+
|
|
|
+ }
|
|
|
+ break;
|
|
|
+
|
|
|
+ default:
|
|
|
+ // TODO what about other states
|
|
|
+ log_printf (LOG_LEVEL_ERROR, "memb_attempt_join: EVS or FORM state attempt join occured %d\n", memb_state);
|
|
|
+ }
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+static int message_handler_memb_join (
|
|
|
+ struct sockaddr_in *system_from,
|
|
|
+ struct iovec *iovec,
|
|
|
+ int iov_len,
|
|
|
+ int bytes_received)
|
|
|
+{
|
|
|
+ struct memb_join *memb_join;
|
|
|
+ int commit_entry;
|
|
|
+ int found;
|
|
|
+ int consensus;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Not representative
|
|
|
+ */
|
|
|
+ if (memb_conf_id.rep.s_addr != memb_local_sockaddr_in.sin_addr.s_addr) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "not the rep for this ring, not handling join.\n");
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+ switch (memb_state) {
|
|
|
+ case MEMB_STATE_OPERATIONAL:
|
|
|
+ case MEMB_STATE_GATHER:
|
|
|
+ memb_state_commit_enter ();
|
|
|
+ /*
|
|
|
+ * do not place break in this case, immediately enter COMMIT state
|
|
|
+ */
|
|
|
+
|
|
|
+ case MEMB_STATE_COMMIT:
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "JOIN in commit\n");
|
|
|
+ memb_join = (struct memb_join *)iovec[0].iov_base;
|
|
|
+ /*
|
|
|
+ * Find gather set that matches the system message was from
|
|
|
+ */
|
|
|
+ for (found = 0, commit_entry = 0; commit_entry < memb_commit_set_entries; commit_entry++) {
|
|
|
+ if (system_from->sin_addr.s_addr == memb_commit_set[commit_entry].rep.sin_addr.s_addr) {
|
|
|
+ found = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Add system from to commit sets if not currently in commit set
|
|
|
+ */
|
|
|
+ if (found == 0) {
|
|
|
+ memcpy (&memb_commit_set[commit_entry].rep, system_from, sizeof (struct sockaddr_in));
|
|
|
+ memb_commit_set_entries++;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Set gather join data
|
|
|
+ */
|
|
|
+ memcpy (memb_commit_set[commit_entry].join_rep_list, memb_join->active_rep_list,
|
|
|
+ sizeof (struct in_addr) * memb_join->active_rep_list_entries);
|
|
|
+ memb_commit_set[commit_entry].join_rep_list_entries = memb_join->active_rep_list_entries;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Union all entries into the gather set (join_rep_list[0])
|
|
|
+ */
|
|
|
+ memb_state_commit_union (commit_entry);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Send JOIN message, but only if gather set has changed
|
|
|
+ */
|
|
|
+ memb_join_send ();
|
|
|
+
|
|
|
+ /*
|
|
|
+ * If consensus, transition to FORM
|
|
|
+ */
|
|
|
+ memb_print_commit_set ();
|
|
|
+
|
|
|
+ consensus = memb_state_consensus_commit ();
|
|
|
+ if (consensus) {
|
|
|
+ log_printf (LOG_LEVEL_NOTICE, "CONSENSUS reached!\n");
|
|
|
+ if (memb_local_sockaddr_in.sin_addr.s_addr == memb_gather_set[0].s_addr) {
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "This node responsible for sending the FORM token.\n");
|
|
|
+
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_memb_state_commit_timeout);
|
|
|
+ timer_memb_state_commit_timeout = 0;
|
|
|
+
|
|
|
+ memb_form_token_send_initial ();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ /*
|
|
|
+ * All other cases are ignored on JOINs
|
|
|
+ */
|
|
|
+ case MEMB_STATE_FORM:
|
|
|
+ log_printf (LOG_LEVEL_WARNING, "JOIN in form, ignoring since consensus reached in state machine.\n");
|
|
|
+ break;
|
|
|
+
|
|
|
+ default:
|
|
|
+ // TODO HANDLE THIS CASE
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "memb_join: DEFAULT case %d, shouldn't happen!!\n", memb_state);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|
|
|
+
|
|
|
+static int message_handler_memb_form_token (
|
|
|
+ struct sockaddr_in *system_from,
|
|
|
+ struct iovec *iovec,
|
|
|
+ int iov_len,
|
|
|
+ int bytes_received)
|
|
|
+{
|
|
|
+ int i;
|
|
|
+ int local = 0;
|
|
|
+ int res = 0;
|
|
|
+ int swallow_form = 0;
|
|
|
+
|
|
|
+printf ("Got membership form token\n");
|
|
|
+ memcpy (&memb_form_token, iovec->iov_base, sizeof (struct memb_form_token));
|
|
|
+
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_form_token_timeout);
|
|
|
+ timer_form_token_timeout = 0;
|
|
|
+
|
|
|
+
|
|
|
+ switch (memb_state) {
|
|
|
+ case MEMB_STATE_OPERATIONAL:
|
|
|
+ case MEMB_STATE_COMMIT:
|
|
|
+ memb_state = MEMB_STATE_FORM;
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_memb_state_commit_timeout);
|
|
|
+ timer_memb_state_commit_timeout = 0;
|
|
|
+ /*
|
|
|
+ * Add member to entry
|
|
|
+ */
|
|
|
+ memb_form_token.member_list[memb_form_token.member_list_entries].s_addr =
|
|
|
+ memb_local_sockaddr_in.sin_addr.s_addr;
|
|
|
+ memb_form_token.member_list_entries++;
|
|
|
+ /*
|
|
|
+ * Modify the conf_id as necessary
|
|
|
+ */
|
|
|
+ memb_form_token_conf_desc_build (&memb_form_token);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Stop token timeout timer from firing
|
|
|
+ * If we are in FORM state, a previous FORM state member
|
|
|
+ * may have captured the ORF token and swallowed it
|
|
|
+ */
|
|
|
+ poll_timer_delete (*gmi_poll_handle, timer_orf_token_timeout);
|
|
|
+ timer_orf_token_timeout = 0;
|
|
|
+ break;
|
|
|
+
|
|
|
+ case MEMB_STATE_FORM:
|
|
|
+ memb_state = MEMB_STATE_EVS;
|
|
|
+ memb_form_token_update_highest_seq (&memb_form_token);
|
|
|
+
|
|
|
+ /*
|
|
|
+ * FORM token has rotated once, now install local variables
|
|
|
+ *
|
|
|
+ * Set barrier sequence number
|
|
|
+ * Set original arut
|
|
|
+ */
|
|
|
+ gmi_barrier_seq = 0;
|
|
|
+printf ("conf_desc_list %d\n", memb_form_token.conf_desc_list_entries);
|
|
|
+ for (i = 0; i < memb_form_token.conf_desc_list_entries; i++) {
|
|
|
+printf ("highest seq %d %d\n", i, memb_form_token.conf_desc_list[i].highest_seq);
|
|
|
+ if (gmi_barrier_seq < memb_form_token.conf_desc_list[i].highest_seq) {
|
|
|
+ gmi_barrier_seq = memb_form_token.conf_desc_list[i].highest_seq;
|
|
|
+printf ("setting barrier seq to %d\n", gmi_barrier_seq);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ gmi_barrier_seq += 1;
|
|
|
+printf ("setting barrier seq to %d\n", gmi_barrier_seq);
|
|
|
+ gmi_original_arut = gmi_arut;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Determine next ORF target
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_form_token.member_list_entries; i++) {
|
|
|
+ if (memb_local_sockaddr_in.sin_addr.s_addr == memb_form_token.member_list[i].s_addr) {
|
|
|
+ memb_next.sin_addr.s_addr =
|
|
|
+ memb_form_token.member_list[(i + 1) % memb_form_token.member_list_entries].s_addr;
|
|
|
+ memb_next.sin_family = AF_INET;
|
|
|
+ memb_next.sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ break;
|
|
|
+
|
|
|
+ case MEMB_STATE_EVS:
|
|
|
+ log_printf (LOG_LEVEL_DEBUG, "Swallowing FORM token in EVS state.\n");
|
|
|
+ printf ("FORM CONF ENTRIES %d\n", memb_form_token.conf_desc_list_entries);
|
|
|
+ orf_token_send_initial();
|
|
|
+ swallow_form = 1;
|
|
|
+ return (0);
|
|
|
+
|
|
|
+ default:
|
|
|
+ // TODO
|
|
|
+ log_printf (LOG_LEVEL_ERROR, "memb_form_token: default case, shouldn't happen.\n");
|
|
|
+ return (0);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Find next member
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_list_entries; i++) {
|
|
|
+ if (memb_list[i].sin_addr.s_addr == memb_local_sockaddr_in.sin_addr.s_addr) {
|
|
|
+ local = 1;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (memb_list_entries == 0) { /* 0 or 1 members and we are local */
|
|
|
+ local = 1;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (local && (i + 1 < memb_list_entries)) {
|
|
|
+ memb_next.sin_addr.s_addr = memb_list[i + 1].sin_addr.s_addr;
|
|
|
+ } else {
|
|
|
+ /*
|
|
|
+ * Find next representative
|
|
|
+ */
|
|
|
+ for (i = 0; i < memb_form_token.rep_list_entries; i++) {
|
|
|
+ if (memb_conf_id.rep.s_addr ==
|
|
|
+ memb_form_token.rep_list[i].s_addr) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ memb_next.sin_addr.s_addr =
|
|
|
+ memb_form_token.rep_list[(i + 1) % memb_form_token.rep_list_entries].s_addr;
|
|
|
+ }
|
|
|
+ memb_next.sin_family = AF_INET;
|
|
|
+ memb_next.sin_port = sockaddr_in_mcast.sin_port;
|
|
|
+ if (swallow_form == 0) {
|
|
|
+ res = memb_form_token_send (&memb_form_token);
|
|
|
+ }
|
|
|
+ return (res);
|
|
|
+}
|
|
|
+
|
|
|
+int recv_handler (poll_handle handle, int fd, int revents, void *data)
|
|
|
+{
|
|
|
+ struct msghdr msg_recv;
|
|
|
+ struct message_header *message_header;
|
|
|
+ struct sockaddr_in system_from;
|
|
|
+ int res = 0;
|
|
|
+ int bytes_received;
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Receive datagram
|
|
|
+ */
|
|
|
+ msg_recv.msg_name = &system_from;
|
|
|
+ msg_recv.msg_namelen = sizeof (struct sockaddr_in);
|
|
|
+ msg_recv.msg_iov = &gmi_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 (gmi_fd, &msg_recv, MSG_NOSIGNAL | MSG_DONTWAIT);
|
|
|
+ if (bytes_received == -1) {
|
|
|
+ return (0);
|
|
|
+ } else {
|
|
|
+ stats_recv += bytes_received;
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Authenticate datagram
|
|
|
+ */
|
|
|
+ res = gmi_msg_auth (msg_recv.msg_iov, msg_recv.msg_iovlen);
|
|
|
+ if (res == -1) {
|
|
|
+ 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;
|
|
|
+ gmi_message_handlers.handler_functions[message_header->type] (
|
|
|
+ &system_from,
|
|
|
+ msg_recv.msg_iov,
|
|
|
+ msg_recv.msg_iovlen,
|
|
|
+ bytes_received);
|
|
|
+
|
|
|
+ return (0);
|
|
|
+}
|