No OneTemporary
Actions

Size

186 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/exec/totemsrp.c b/exec/totemsrp.c
	index e05d65bc..53211bae 100644
	--- a/exec/totemsrp.c
	+++ b/exec/totemsrp.c
	@@ -1,4497 +1,4493 @@
	/*
	* Copyright (c) 2003-2006 MontaVista Software, Inc.
	* Copyright (c) 2006-2009 Red Hat, Inc.
	*
	* All rights reserved.
	*
	* Author: Steven Dake (sdake@redhat.com)
	*
	* This software licensed under BSD license, the text of which follows:
	*
	* Redistribution and use in source and binary forms, with or without
	* modification, are permitted provided that the following conditions are met:
	*
	* - Redistributions of source code must retain the above copyright notice,
	* this list of conditions and the following disclaimer.
	* - Redistributions in binary form must reproduce the above copyright notice,
	* this list of conditions and the following disclaimer in the documentation
	* and/or other materials provided with the distribution.
	* - Neither the name of the MontaVista Software, Inc. nor the names of its
	* contributors may be used to endorse or promote products derived from this
	* software without specific prior written permission.
	*
	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
	* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
	* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
	* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
	* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
	* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
	* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
	* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
	* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
	* THE POSSIBILITY OF SUCH DAMAGE.
	*/

	/*
	* The first version of this code was based upon Yair Amir's PhD thesis:
	* http://www.cs.jhu.edu/~yairamir/phd.ps) (ch4,5).
	*
	* The current version of totemsrp implements the Totem protocol specified in:
	* http://citeseer.ist.psu.edu/amir95totem.html
	*
	* The deviations from the above published protocols are:
	* - encryption of message contents with SOBER128
	* - authentication of meessage contents with SHA1/HMAC
	* - token hold mode where token doesn't rotate on unused ring - reduces cpu
	* usage on 1.6ghz xeon from 35% to less then .1 % as measured by top
	*/

	#include <config.h>

	#include <assert.h>
	#ifdef HAVE_ALLOCA_H
	#include <alloca.h>
	#endif
	#include <sys/mman.h>
	#include <sys/types.h>
	#include <sys/stat.h>
	#include <sys/socket.h>
	#include <netdb.h>
	#include <sys/un.h>
	#include <sys/ioctl.h>
	#include <sys/param.h>
	#include <netinet/in.h>
	#include <arpa/inet.h>
	#include <unistd.h>
	#include <fcntl.h>
	#include <stdlib.h>
	#include <stdio.h>
	#include <errno.h>
	#include <sched.h>
	#include <time.h>
	#include <sys/time.h>
	#include <sys/poll.h>
	#include <limits.h>

	#include <qb/qbdefs.h>
	#include <qb/qbutil.h>
	#include <qb/qbloop.h>

	#include <corosync/swab.h>
	#include <corosync/cs_queue.h>
	#include <corosync/sq.h>
	#include <corosync/list.h>
	#include <corosync/hdb.h>

	#define LOGSYS_UTILS_ONLY 1
	#include <corosync/engine/logsys.h>

	#include "totemsrp.h"
	#include "totemrrp.h"
	#include "totemnet.h"

	#include "crypto.h"

	#define LOCALHOST_IP inet_addr("127.0.0.1")
	#define QUEUE_RTR_ITEMS_SIZE_MAX 16384 /* allow 16384 retransmit items */
	#define RETRANS_MESSAGE_QUEUE_SIZE_MAX 16384 /* allow 500 messages to be queued */
	#define RECEIVED_MESSAGE_QUEUE_SIZE_MAX 500 /* allow 500 messages to be queued */
	#define MAXIOVS 5
	#define RETRANSMIT_ENTRIES_MAX 30
	#define TOKEN_SIZE_MAX 64000 /* bytes */
	#define LEAVE_DUMMY_NODEID 0

	/*
	* Rollover handling:
	* SEQNO_START_MSG is the starting sequence number after a new configuration
	* This should remain zero, unless testing overflow in which case
	* 0x7ffff000 and 0xfffff000 are good starting values.
	*
	* SEQNO_START_TOKEN is the starting sequence number after a new configuration
	* for a token. This should remain zero, unless testing overflow in which
	* case 07fffff00 or 0xffffff00 are good starting values.
	- *
	- * SEQNO_START_MSG is the starting sequence number after a new configuration
	- * This should remain zero, unless testing overflow in which case
	- * 0x7ffff000 and 0xfffff000 are good values to start with
	*/
	#define SEQNO_START_MSG 0x0
	#define SEQNO_START_TOKEN 0x0

	/*
	* These can be used ot test different rollover points
	* #define SEQNO_START_MSG 0xfffffe00
	* #define SEQNO_START_TOKEN 0xfffffe00
	*/

	/*
	* These can be used to test the error recovery algorithms
	* #define TEST_DROP_ORF_TOKEN_PERCENTAGE 30
	* #define TEST_DROP_COMMIT_TOKEN_PERCENTAGE 30
	* #define TEST_DROP_MCAST_PERCENTAGE 50
	* #define TEST_RECOVERY_MSG_COUNT 300
	*/

	/*
	* we compare incoming messages to determine if their endian is
	* different - if so convert them
	*
	* do not change
	*/
	#define ENDIAN_LOCAL 0xff22

	enum message_type {
	MESSAGE_TYPE_ORF_TOKEN = 0, /* Ordering, Reliability, Flow (ORF) control Token */
	MESSAGE_TYPE_MCAST = 1, /* ring ordered multicast message */
	MESSAGE_TYPE_MEMB_MERGE_DETECT = 2, /* merge rings if there are available rings */
	MESSAGE_TYPE_MEMB_JOIN = 3, /* membership join message */
	MESSAGE_TYPE_MEMB_COMMIT_TOKEN = 4, /* membership commit token */
	MESSAGE_TYPE_TOKEN_HOLD_CANCEL = 5, /* cancel the holding of the token */
	};

	enum encapsulation_type {
	MESSAGE_ENCAPSULATED = 1,
	MESSAGE_NOT_ENCAPSULATED = 2
	};

	/*
	* New membership algorithm local variables
	*/
	struct srp_addr {
	struct totem_ip_address addr[INTERFACE_MAX];
	};


	struct consensus_list_item {
	struct srp_addr addr;
	int set;
	};


	struct token_callback_instance {
	struct list_head list;
	int (callback_fn) (enum totem_callback_token_type type, const void );
	enum totem_callback_token_type callback_type;
	int delete;
	void *data;
	};


	struct totemsrp_socket {
	int mcast;
	int token;
	};

	struct message_header {
	char type;
	char encapsulated;
	unsigned short endian_detector;
	unsigned int nodeid;
	} __attribute__((packed));


	struct mcast {
	struct message_header header;
	struct srp_addr system_from;
	unsigned int seq;
	int this_seqno;
	struct memb_ring_id ring_id;
	unsigned int node_id;
	int guarantee;
	} __attribute__((packed));


	struct rtr_item {
	struct memb_ring_id ring_id;
	unsigned int seq;
	}__attribute__((packed));


	struct orf_token {
	struct message_header header;
	unsigned int seq;
	unsigned int token_seq;
	unsigned int aru;
	unsigned int aru_addr;
	struct memb_ring_id ring_id;
	unsigned int backlog;
	unsigned int fcc;
	int retrans_flg;
	int rtr_list_entries;
	struct rtr_item rtr_list[0];
	}__attribute__((packed));


	struct memb_join {
	struct message_header header;
	struct srp_addr system_from;
	unsigned int proc_list_entries;
	unsigned int failed_list_entries;
	unsigned long long ring_seq;
	unsigned char end_of_memb_join[0];
	/*
	* These parts of the data structure are dynamic:
	* struct srp_addr proc_list[];
	* struct srp_addr failed_list[];
	*/
	} __attribute__((packed));


	struct memb_merge_detect {
	struct message_header header;
	struct srp_addr system_from;
	struct memb_ring_id ring_id;
	} __attribute__((packed));


	struct token_hold_cancel {
	struct message_header header;
	struct memb_ring_id ring_id;
	} __attribute__((packed));


	struct memb_commit_token_memb_entry {
	struct memb_ring_id ring_id;
	unsigned int aru;
	unsigned int high_delivered;
	unsigned int received_flg;
	}__attribute__((packed));


	struct memb_commit_token {
	struct message_header header;
	unsigned int token_seq;
	struct memb_ring_id ring_id;
	unsigned int retrans_flg;
	int memb_index;
	int addr_entries;
	unsigned char end_of_commit_token[0];
	/*
	* These parts of the data structure are dynamic:
	*
	* struct srp_addr addr[PROCESSOR_COUNT_MAX];
	* struct memb_commit_token_memb_entry memb_list[PROCESSOR_COUNT_MAX];
	*/
	}__attribute__((packed));

	struct message_item {
	struct mcast *mcast;
	unsigned int msg_len;
	};

	struct sort_queue_item {
	struct mcast *mcast;
	unsigned int msg_len;
	};

	struct orf_token_mcast_thread_state {
	char iobuf[9000];
	prng_state prng_state;
	};

	enum memb_state {
	MEMB_STATE_OPERATIONAL = 1,
	MEMB_STATE_GATHER = 2,
	MEMB_STATE_COMMIT = 3,
	MEMB_STATE_RECOVERY = 4
	};

	struct totemsrp_instance {
	int iface_changes;

	int failed_to_recv;

	/*
	* Flow control mcasts and remcasts on last and current orf_token
	*/
	int fcc_remcast_last;

	int fcc_mcast_last;

	int fcc_remcast_current;

	struct consensus_list_item consensus_list[PROCESSOR_COUNT_MAX];

	int consensus_list_entries;

	struct srp_addr my_id;

	struct srp_addr my_proc_list[PROCESSOR_COUNT_MAX];

	struct srp_addr my_failed_list[PROCESSOR_COUNT_MAX];

	struct srp_addr my_new_memb_list[PROCESSOR_COUNT_MAX];

	struct srp_addr my_trans_memb_list[PROCESSOR_COUNT_MAX];

	struct srp_addr my_memb_list[PROCESSOR_COUNT_MAX];

	struct srp_addr my_deliver_memb_list[PROCESSOR_COUNT_MAX];

	struct srp_addr my_left_memb_list[PROCESSOR_COUNT_MAX];

	int my_proc_list_entries;

	int my_failed_list_entries;

	int my_new_memb_entries;

	int my_trans_memb_entries;

	int my_memb_entries;

	int my_deliver_memb_entries;

	int my_left_memb_entries;

	struct memb_ring_id my_ring_id;

	struct memb_ring_id my_old_ring_id;

	int my_aru_count;

	int my_merge_detect_timeout_outstanding;

	unsigned int my_last_aru;

	int my_seq_unchanged;

	int my_received_flg;

	unsigned int my_high_seq_received;

	unsigned int my_install_seq;

	int my_rotation_counter;

	int my_set_retrans_flg;

	int my_retrans_flg_count;

	unsigned int my_high_ring_delivered;

	int heartbeat_timeout;

	/*
	* Queues used to order, deliver, and recover messages
	*/
	struct cs_queue new_message_queue;

	struct cs_queue retrans_message_queue;

	struct sq regular_sort_queue;

	struct sq recovery_sort_queue;

	/*
	* Received up to and including
	*/
	unsigned int my_aru;

	unsigned int my_high_delivered;

	struct list_head token_callback_received_listhead;

	struct list_head token_callback_sent_listhead;

	char orf_token_retransmit[TOKEN_SIZE_MAX];

	int orf_token_retransmit_size;

	unsigned int my_token_seq;

	/*
	* Timers
	*/
	qb_loop_timer_handle timer_pause_timeout;

	qb_loop_timer_handle timer_orf_token_timeout;

	qb_loop_timer_handle timer_orf_token_retransmit_timeout;

	qb_loop_timer_handle timer_orf_token_hold_retransmit_timeout;

	qb_loop_timer_handle timer_merge_detect_timeout;

	qb_loop_timer_handle memb_timer_state_gather_join_timeout;

	qb_loop_timer_handle memb_timer_state_gather_consensus_timeout;

	qb_loop_timer_handle memb_timer_state_commit_timeout;

	qb_loop_timer_handle timer_heartbeat_timeout;

	/*
	* Function and data used to log messages
	*/
	int totemsrp_log_level_security;

	int totemsrp_log_level_error;

	int totemsrp_log_level_warning;

	int totemsrp_log_level_notice;

	int totemsrp_log_level_debug;

	int totemsrp_subsys_id;

	void (*totemsrp_log_printf) (
	int level,
	int sybsys,
	const char *function,
	const char *file,
	int line,
	const char *format, ...)__attribute__((format(printf, 6, 7)));;

	enum memb_state memb_state;

	//TODO struct srp_addr next_memb;

	qb_loop_t *totemsrp_poll_handle;

	struct totem_ip_address mcast_address;

	void (*totemsrp_deliver_fn) (
	unsigned int nodeid,
	const void *msg,
	unsigned int msg_len,
	int endian_conversion_required);

	void (*totemsrp_confchg_fn) (
	enum totem_configuration_type configuration_type,
	const unsigned int *member_list, size_t member_list_entries,
	const unsigned int *left_list, size_t left_list_entries,
	const unsigned int *joined_list, size_t joined_list_entries,
	const struct memb_ring_id *ring_id);

	void (*totemsrp_service_ready_fn) (void);

	int global_seqno;

	int my_token_held;

	unsigned long long token_ring_id_seq;

	unsigned int last_released;

	unsigned int set_aru;

	int old_ring_state_saved;

	int old_ring_state_aru;

	unsigned int old_ring_state_high_seq_received;

	unsigned int my_last_seq;

	struct timeval tv_old;

	void *totemrrp_context;

	struct totem_config *totem_config;

	unsigned int use_heartbeat;

	unsigned int my_trc;

	unsigned int my_pbl;

	unsigned int my_cbl;

	uint64_t pause_timestamp;

	struct memb_commit_token *commit_token;

	totemsrp_stats_t stats;

	uint32_t orf_token_discard;

	void * token_recv_event_handle;
	void * token_sent_event_handle;
	char commit_token_storage[40000];
	};

	struct message_handlers {
	int count;
	int (*handler_functions[6]) (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed);
	};

	/*
	* forward decls
	*/
	static int message_handler_orf_token (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed);

	static int message_handler_mcast (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed);

	static int message_handler_memb_merge_detect (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed);

	static int message_handler_memb_join (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed);

	static int message_handler_memb_commit_token (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed);

	static int message_handler_token_hold_cancel (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed);

	static void totemsrp_instance_initialize (struct totemsrp_instance *instance);

	static unsigned int main_msgs_missing (void);

	static void main_token_seqid_get (
	const void *msg,
	unsigned int *seqid,
	unsigned int *token_is);

	static void srp_addr_copy (struct srp_addr dest, const struct srp_addr src);

	static void srp_addr_to_nodeid (
	unsigned int *nodeid_out,
	struct srp_addr *srp_addr_in,
	unsigned int entries);

	static int srp_addr_equal (const struct srp_addr a, const struct srp_addr b);

	static void memb_leave_message_send (struct totemsrp_instance *instance);

	static void memb_ring_id_create_or_load (struct totemsrp_instance , struct memb_ring_id );

	static void token_callbacks_execute (struct totemsrp_instance *instance, enum totem_callback_token_type type);
	static void memb_state_gather_enter (struct totemsrp_instance *instance, int gather_from);
	static void messages_deliver_to_app (struct totemsrp_instance *instance, int skip, unsigned int end_point);
	static int orf_token_mcast (struct totemsrp_instance instance, struct orf_token oken,
	int fcc_mcasts_allowed);
	static void messages_free (struct totemsrp_instance *instance, unsigned int token_aru);

	static void memb_ring_id_set_and_store (struct totemsrp_instance *instance,
	const struct memb_ring_id *ring_id);
	static void target_set_completed (void *context);
	static void memb_state_commit_token_update (struct totemsrp_instance *instance);
	static void memb_state_commit_token_target_set (struct totemsrp_instance *instance);
	static int memb_state_commit_token_send (struct totemsrp_instance *instance);
	static int memb_state_commit_token_send_recovery (struct totemsrp_instance instance, struct memb_commit_token memb_commit_token);
	static void memb_state_commit_token_create (struct totemsrp_instance *instance);
	static int token_hold_cancel_send (struct totemsrp_instance *instance);
	static void orf_token_endian_convert (const struct orf_token in, struct orf_token out);
	static void memb_commit_token_endian_convert (const struct memb_commit_token in, struct memb_commit_token out);
	static void memb_join_endian_convert (const struct memb_join in, struct memb_join out);
	static void mcast_endian_convert (const struct mcast in, struct mcast out);
	static void memb_merge_detect_endian_convert (
	const struct memb_merge_detect *in,
	struct memb_merge_detect *out);
	static void srp_addr_copy_endian_convert (struct srp_addr out, const struct srp_addr in);
	static void timer_function_orf_token_timeout (void *data);
	static void timer_function_pause_timeout (void *data);
	static void timer_function_heartbeat_timeout (void *data);
	static void timer_function_token_retransmit_timeout (void *data);
	static void timer_function_token_hold_retransmit_timeout (void *data);
	static void timer_function_merge_detect_timeout (void *data);
	static void totemsrp_buffer_alloc (struct totemsrp_instance instance);
	static void totemsrp_buffer_release (struct totemsrp_instance instance, void ptr);

	void main_deliver_fn (
	void *context,
	const void *msg,
	unsigned int msg_len);

	void main_iface_change_fn (
	void *context,
	const struct totem_ip_address *iface_address,
	unsigned int iface_no);

	struct message_handlers totemsrp_message_handlers = {
	6,
	{
	- message_handler_orf_token,
	- message_handler_mcast,
	- message_handler_memb_merge_detect,
	- message_handler_memb_join,
	- message_handler_memb_commit_token,
	- message_handler_token_hold_cancel
	+ message_handler_orf_token, /* MESSAGE_TYPE_ORF_TOKEN */
	+ message_handler_mcast, /* MESSAGE_TYPE_MCAST */
	+ message_handler_memb_merge_detect, /* MESSAGE_TYPE_MEMB_MERGE_DETECT */
	+ message_handler_memb_join, /* MESSAGE_TYPE_MEMB_JOIN */
	+ message_handler_memb_commit_token, /* MESSAGE_TYPE_MEMB_COMMIT_TOKEN */
	+ message_handler_token_hold_cancel /* MESSAGE_TYPE_TOKEN_HOLD_CANCEL */
	}
	};

	static const char *rundir = NULL;

	#define log_printf(level, format, args...) \
	do { \
	instance->totemsrp_log_printf ( \
	level, instance->totemsrp_subsys_id, \
	__FUNCTION__, __FILE__, __LINE__, \
	format, ##args); \
	} while (0);
	#define LOGSYS_PERROR(err_num, level, fmt, args...) \
	do { \
	char _error_str[LOGSYS_MAX_PERROR_MSG_LEN]; \
	const char *_error_ptr = qb_strerror_r(err_num, _error_str, sizeof(_error_str)); \
	instance->totemsrp_log_printf ( \
	level, instance->totemsrp_subsys_id, \
	__FUNCTION__, __FILE__, __LINE__, \
	fmt ": %s (%d)\n", ##args, _error_ptr, err_num); \
	} while(0)

	static void totemsrp_instance_initialize (struct totemsrp_instance *instance)
	{
	memset (instance, 0, sizeof (struct totemsrp_instance));

	list_init (&instance->token_callback_received_listhead);

	list_init (&instance->token_callback_sent_listhead);

	instance->my_received_flg = 1;

	instance->my_token_seq = SEQNO_START_TOKEN - 1;

	instance->memb_state = MEMB_STATE_OPERATIONAL;

	instance->set_aru = -1;

	instance->my_aru = SEQNO_START_MSG;

	instance->my_high_seq_received = SEQNO_START_MSG;

	instance->my_high_delivered = SEQNO_START_MSG;

	instance->orf_token_discard = 0;

	instance->commit_token = (struct memb_commit_token *)instance->commit_token_storage;
	}

	static void main_token_seqid_get (
	const void *msg,
	unsigned int *seqid,
	unsigned int *token_is)
	{
	const struct orf_token *token = msg;

	*seqid = 0;
	*token_is = 0;
	if (token->header.type == MESSAGE_TYPE_ORF_TOKEN) {
	*seqid = token->token_seq;
	*token_is = 1;
	}
	}

	static unsigned int main_msgs_missing (void)
	{
	// TODO
	return (0);
	}

	static int pause_flush (struct totemsrp_instance *instance)
	{
	uint64_t now_msec;
	uint64_t timestamp_msec;
	int res = 0;

	now_msec = (qb_util_nano_current_get () / QB_TIME_NS_IN_MSEC);
	timestamp_msec = instance->pause_timestamp / QB_TIME_NS_IN_MSEC;

	if ((now_msec - timestamp_msec) > (instance->totem_config->token_timeout / 2)) {
	log_printf (instance->totemsrp_log_level_notice,
	"Process pause detected for %d ms, flushing membership messages.\n", (unsigned int)(now_msec - timestamp_msec));
	/*
	* -1 indicates an error from recvmsg
	*/
	do {
	res = totemrrp_mcast_recv_empty (instance->totemrrp_context);
	} while (res == -1);
	}
	return (res);
	}

	static int token_event_stats_collector (enum totem_callback_token_type type, const void *void_instance)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )void_instance;
	uint32_t time_now;
	unsigned long long nano_secs = qb_util_nano_current_get ();

	time_now = (nano_secs / QB_TIME_NS_IN_MSEC);

	if (type == TOTEM_CALLBACK_TOKEN_RECEIVED) {
	/* incr latest token the index */
	if (instance->stats.latest_token == (TOTEM_TOKEN_STATS_MAX - 1))
	instance->stats.latest_token = 0;
	else
	instance->stats.latest_token++;

	if (instance->stats.earliest_token == instance->stats.latest_token) {
	/* we have filled up the array, start overwriting */
	if (instance->stats.earliest_token == (TOTEM_TOKEN_STATS_MAX - 1))
	instance->stats.earliest_token = 0;
	else
	instance->stats.earliest_token++;

	instance->stats.token[instance->stats.earliest_token].rx = 0;
	instance->stats.token[instance->stats.earliest_token].tx = 0;
	instance->stats.token[instance->stats.earliest_token].backlog_calc = 0;
	}

	instance->stats.token[instance->stats.latest_token].rx = time_now;
	instance->stats.token[instance->stats.latest_token].tx = 0; /* in case we drop the token */
	} else {
	instance->stats.token[instance->stats.latest_token].tx = time_now;
	}
	return 0;
	}

	/*
	* Exported interfaces
	*/
	int totemsrp_initialize (
	qb_loop_t *poll_handle,
	void **srp_context,
	struct totem_config *totem_config,
	totemmrp_stats_t *stats,

	void (*deliver_fn) (
	unsigned int nodeid,
	const void *msg,
	unsigned int msg_len,
	int endian_conversion_required),

	void (*confchg_fn) (
	enum totem_configuration_type configuration_type,
	const unsigned int *member_list, size_t member_list_entries,
	const unsigned int *left_list, size_t left_list_entries,
	const unsigned int *joined_list, size_t joined_list_entries,
	const struct memb_ring_id *ring_id))
	{
	struct totemsrp_instance *instance;
	unsigned int res;

	instance = malloc (sizeof (struct totemsrp_instance));
	if (instance == NULL) {
	goto error_exit;
	}

	rundir = getenv ("COROSYNC_RUN_DIR");
	if (rundir == NULL) {
	rundir = LOCALSTATEDIR "/lib/corosync";
	}

	res = mkdir (rundir, 0700);
	if (res == -1 && errno != EEXIST) {
	goto error_destroy;
	}

	res = chdir (rundir);
	if (res == -1) {
	goto error_destroy;
	}

	totemsrp_instance_initialize (instance);

	stats->srp = &instance->stats;
	instance->stats.latest_token = 0;
	instance->stats.earliest_token = 0;

	instance->totem_config = totem_config;

	/*
	* Configure logging
	*/
	instance->totemsrp_log_level_security = totem_config->totem_logging_configuration.log_level_security;
	instance->totemsrp_log_level_error = totem_config->totem_logging_configuration.log_level_error;
	instance->totemsrp_log_level_warning = totem_config->totem_logging_configuration.log_level_warning;
	instance->totemsrp_log_level_notice = totem_config->totem_logging_configuration.log_level_notice;
	instance->totemsrp_log_level_debug = totem_config->totem_logging_configuration.log_level_debug;
	instance->totemsrp_subsys_id = totem_config->totem_logging_configuration.log_subsys_id;
	instance->totemsrp_log_printf = totem_config->totem_logging_configuration.log_printf;

	/*
	* Initialize local variables for totemsrp
	*/
	totemip_copy (&instance->mcast_address, &totem_config->interfaces[0].mcast_addr);

	/*
	* Display totem configuration
	*/
	log_printf (instance->totemsrp_log_level_debug,
	"Token Timeout (%d ms) retransmit timeout (%d ms)\n",
	totem_config->token_timeout, totem_config->token_retransmit_timeout);
	log_printf (instance->totemsrp_log_level_debug,
	"token hold (%d ms) retransmits before loss (%d retrans)\n",
	totem_config->token_hold_timeout, totem_config->token_retransmits_before_loss_const);
	log_printf (instance->totemsrp_log_level_debug,
	"join (%d ms) send_join (%d ms) consensus (%d ms) merge (%d ms)\n",
	totem_config->join_timeout,
	totem_config->send_join_timeout,
	totem_config->consensus_timeout,

	totem_config->merge_timeout);
	log_printf (instance->totemsrp_log_level_debug,
	"downcheck (%d ms) fail to recv const (%d msgs)\n",
	totem_config->downcheck_timeout, totem_config->fail_to_recv_const);
	log_printf (instance->totemsrp_log_level_debug,
	"seqno unchanged const (%d rotations) Maximum network MTU %d\n", totem_config->seqno_unchanged_const, totem_config->net_mtu);

	log_printf (instance->totemsrp_log_level_debug,
	"window size per rotation (%d messages) maximum messages per rotation (%d messages)\n",
	totem_config->window_size, totem_config->max_messages);

	log_printf (instance->totemsrp_log_level_debug,
	"missed count const (%d messages)\n",
	totem_config->miss_count_const);

	log_printf (instance->totemsrp_log_level_debug,
	"send threads (%d threads)\n", totem_config->threads);
	log_printf (instance->totemsrp_log_level_debug,
	"RRP token expired timeout (%d ms)\n",
	totem_config->rrp_token_expired_timeout);
	log_printf (instance->totemsrp_log_level_debug,
	"RRP token problem counter (%d ms)\n",
	totem_config->rrp_problem_count_timeout);
	log_printf (instance->totemsrp_log_level_debug,
	"RRP threshold (%d problem count)\n",
	totem_config->rrp_problem_count_threshold);
	log_printf (instance->totemsrp_log_level_debug,
	"RRP automatic recovery check timeout (%d ms)\n",
	totem_config->rrp_autorecovery_check_timeout);
	log_printf (instance->totemsrp_log_level_debug,
	"RRP mode set to %s.\n", instance->totem_config->rrp_mode);

	log_printf (instance->totemsrp_log_level_debug,
	"heartbeat_failures_allowed (%d)\n", totem_config->heartbeat_failures_allowed);
	log_printf (instance->totemsrp_log_level_debug,
	"max_network_delay (%d ms)\n", totem_config->max_network_delay);


	cs_queue_init (&instance->retrans_message_queue, RETRANS_MESSAGE_QUEUE_SIZE_MAX,
	sizeof (struct message_item));

	sq_init (&instance->regular_sort_queue,
	QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0);

	sq_init (&instance->recovery_sort_queue,
	QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0);

	instance->totemsrp_poll_handle = poll_handle;

	instance->totemsrp_deliver_fn = deliver_fn;

	instance->totemsrp_confchg_fn = confchg_fn;
	instance->use_heartbeat = 1;

	timer_function_pause_timeout (instance);

	if ( totem_config->heartbeat_failures_allowed == 0 ) {
	log_printf (instance->totemsrp_log_level_debug,
	"HeartBeat is Disabled. To enable set heartbeat_failures_allowed > 0\n");
	instance->use_heartbeat = 0;
	}

	if (instance->use_heartbeat) {
	instance->heartbeat_timeout
	= (totem_config->heartbeat_failures_allowed) * totem_config->token_retransmit_timeout
	+ totem_config->max_network_delay;

	if (instance->heartbeat_timeout >= totem_config->token_timeout) {
	log_printf (instance->totemsrp_log_level_debug,
	"total heartbeat_timeout (%d ms) is not less than token timeout (%d ms)\n",
	instance->heartbeat_timeout,
	totem_config->token_timeout);
	log_printf (instance->totemsrp_log_level_debug,
	"heartbeat_timeout = heartbeat_failures_allowed * token_retransmit_timeout + max_network_delay\n");
	log_printf (instance->totemsrp_log_level_debug,
	"heartbeat timeout should be less than the token timeout. HeartBeat is Diabled !!\n");
	instance->use_heartbeat = 0;
	}
	else {
	log_printf (instance->totemsrp_log_level_debug,
	"total heartbeat_timeout (%d ms)\n", instance->heartbeat_timeout);
	}
	}

	totemrrp_initialize (
	poll_handle,
	&instance->totemrrp_context,
	totem_config,
	instance,
	main_deliver_fn,
	main_iface_change_fn,
	main_token_seqid_get,
	main_msgs_missing,
	target_set_completed);

	/*
	* Must have net_mtu adjusted by totemrrp_initialize first
	*/
	cs_queue_init (&instance->new_message_queue,
	MESSAGE_QUEUE_MAX,
	sizeof (struct message_item));

	totemsrp_callback_token_create (instance,
	&instance->token_recv_event_handle,
	TOTEM_CALLBACK_TOKEN_RECEIVED,
	0,
	token_event_stats_collector,
	instance);
	totemsrp_callback_token_create (instance,
	&instance->token_sent_event_handle,
	TOTEM_CALLBACK_TOKEN_SENT,
	0,
	token_event_stats_collector,
	instance);
	*srp_context = instance;
	return (0);

	error_destroy:
	free (instance);

	error_exit:
	return (-1);
	}

	void totemsrp_finalize (
	void *srp_context)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;

	memb_leave_message_send (instance);
	free (srp_context);
	}

	int totemsrp_ifaces_get (
	void *srp_context,
	unsigned int nodeid,
	struct totem_ip_address *interfaces,
	char ***status,
	unsigned int *iface_count)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;
	int res = 0;
	unsigned int found = 0;
	unsigned int i;

	for (i = 0; i < instance->my_memb_entries; i++) {
	if (instance->my_memb_list[i].addr[0].nodeid == nodeid) {
	found = 1;
	break;
	}
	}

	if (found) {
	memcpy (interfaces, &instance->my_memb_list[i],
	sizeof (struct srp_addr));
	*iface_count = instance->totem_config->interface_count;
	goto finish;
	}

	for (i = 0; i < instance->my_left_memb_entries; i++) {
	if (instance->my_left_memb_list[i].addr[0].nodeid == nodeid) {
	found = 1;
	break;
	}
	}

	if (found) {
	memcpy (interfaces, &instance->my_left_memb_list[i],
	sizeof (struct srp_addr));
	*iface_count = instance->totem_config->interface_count;
	} else {
	res = -1;
	}

	finish:
	totemrrp_ifaces_get (instance->totemrrp_context, status, NULL);
	return (res);
	}

	int totemsrp_crypto_set (
	void *srp_context,
	unsigned int type)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;
	int res;

	res = totemrrp_crypto_set(instance->totemrrp_context, type);

	return (res);
	}


	unsigned int totemsrp_my_nodeid_get (
	void *srp_context)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;
	unsigned int res;

	res = instance->totem_config->interfaces[0].boundto.nodeid;

	return (res);
	}

	int totemsrp_my_family_get (
	void *srp_context)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;
	int res;

	res = instance->totem_config->interfaces[0].boundto.family;

	return (res);
	}


	int totemsrp_ring_reenable (
	void *srp_context)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;

	totemrrp_ring_reenable (instance->totemrrp_context,
	instance->totem_config->interface_count);

	return (0);
	}


	/*
	* Set operations for use by the membership algorithm
	*/
	static int srp_addr_equal (const struct srp_addr a, const struct srp_addr b)
	{
	unsigned int i;
	unsigned int res;

	for (i = 0; i < 1; i++) {
	res = totemip_equal (&a->addr[i], &b->addr[i]);
	if (res == 0) {
	return (0);
	}
	}
	return (1);
	}

	static void srp_addr_copy (struct srp_addr dest, const struct srp_addr src)
	{
	unsigned int i;

	for (i = 0; i < INTERFACE_MAX; i++) {
	totemip_copy (&dest->addr[i], &src->addr[i]);
	}
	}

	static void srp_addr_to_nodeid (
	unsigned int *nodeid_out,
	struct srp_addr *srp_addr_in,
	unsigned int entries)
	{
	unsigned int i;

	for (i = 0; i < entries; i++) {
	nodeid_out[i] = srp_addr_in[i].addr[0].nodeid;
	}
	}

	static void srp_addr_copy_endian_convert (struct srp_addr out, const struct srp_addr in)
	{
	int i;

	for (i = 0; i < INTERFACE_MAX; i++) {
	totemip_copy_endian_convert (&out->addr[i], &in->addr[i]);
	}
	}

	static void memb_consensus_reset (struct totemsrp_instance *instance)
	{
	instance->consensus_list_entries = 0;
	}

	static void memb_set_subtract (
	struct srp_addr out_list, int out_list_entries,
	struct srp_addr *one_list, int one_list_entries,
	struct srp_addr *two_list, int two_list_entries)
	{
	int found = 0;
	int i;
	int j;

	*out_list_entries = 0;

	for (i = 0; i < one_list_entries; i++) {
	for (j = 0; j < two_list_entries; j++) {
	if (srp_addr_equal (&one_list[i], &two_list[j])) {
	found = 1;
	break;
	}
	}
	if (found == 0) {
	srp_addr_copy (&out_list[*out_list_entries], &one_list[i]);
	out_list_entries = out_list_entries + 1;
	}
	found = 0;
	}
	}

	/*
	* Set consensus for a specific processor
	*/
	static void memb_consensus_set (
	struct totemsrp_instance *instance,
	const struct srp_addr *addr)
	{
	int found = 0;
	int i;

	if (addr->addr[0].nodeid == LEAVE_DUMMY_NODEID)
	return;

	for (i = 0; i < instance->consensus_list_entries; i++) {
	if (srp_addr_equal(addr, &instance->consensus_list[i].addr)) {
	found = 1;
	break; /* found entry */
	}
	}
	srp_addr_copy (&instance->consensus_list[i].addr, addr);
	instance->consensus_list[i].set = 1;
	if (found == 0) {
	instance->consensus_list_entries++;
	}
	return;
	}

	/*
	* Is consensus set for a specific processor
	*/
	static int memb_consensus_isset (
	struct totemsrp_instance *instance,
	const struct srp_addr *addr)
	{
	int i;

	for (i = 0; i < instance->consensus_list_entries; i++) {
	if (srp_addr_equal (addr, &instance->consensus_list[i].addr)) {
	return (instance->consensus_list[i].set);
	}
	}
	return (0);
	}

	/*
	* Is consensus agreed upon based upon consensus database
	*/
	static int memb_consensus_agreed (
	struct totemsrp_instance *instance)
	{
	struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
	int token_memb_entries = 0;
	int agreed = 1;
	int i;

	memb_set_subtract (token_memb, &token_memb_entries,
	instance->my_proc_list, instance->my_proc_list_entries,
	instance->my_failed_list, instance->my_failed_list_entries);

	for (i = 0; i < token_memb_entries; i++) {
	if (memb_consensus_isset (instance, &token_memb[i]) == 0) {
	agreed = 0;
	break;
	}
	}
	assert (token_memb_entries >= 1);

	return (agreed);
	}

	static void memb_consensus_notset (
	struct totemsrp_instance *instance,
	struct srp_addr *no_consensus_list,
	int *no_consensus_list_entries,
	struct srp_addr *comparison_list,
	int comparison_list_entries)
	{
	int i;

	*no_consensus_list_entries = 0;

	for (i = 0; i < instance->my_proc_list_entries; i++) {
	if (memb_consensus_isset (instance, &instance->my_proc_list[i]) == 0) {
	srp_addr_copy (&no_consensus_list[*no_consensus_list_entries], &instance->my_proc_list[i]);
	no_consensus_list_entries = no_consensus_list_entries + 1;
	}
	}
	}

	/*
	* Is set1 equal to set2 Entries can be in different orders
	*/
	static int memb_set_equal (
	struct srp_addr *set1, int set1_entries,
	struct srp_addr *set2, int set2_entries)
	{
	int i;
	int j;

	int found = 0;

	if (set1_entries != set2_entries) {
	return (0);
	}
	for (i = 0; i < set2_entries; i++) {
	for (j = 0; j < set1_entries; j++) {
	if (srp_addr_equal (&set1[j], &set2[i])) {
	found = 1;
	break;
	}
	}
	if (found == 0) {
	return (0);
	}
	found = 0;
	}
	return (1);
	}

	/*
	* Is subset fully contained in fullset
	*/
	static int memb_set_subset (
	const struct srp_addr *subset, int subset_entries,
	const struct srp_addr *fullset, int fullset_entries)
	{
	int i;
	int j;
	int found = 0;

	if (subset_entries > fullset_entries) {
	return (0);
	}
	for (i = 0; i < subset_entries; i++) {
	for (j = 0; j < fullset_entries; j++) {
	if (srp_addr_equal (&subset[i], &fullset[j])) {
	found = 1;
	}
	}
	if (found == 0) {
	return (0);
	}
	found = 0;
	}
	return (1);
	}
	/*
	* merge subset into fullset taking care not to add duplicates
	*/
	static void memb_set_merge (
	const struct srp_addr *subset, int subset_entries,
	struct srp_addr fullset, int fullset_entries)
	{
	int found = 0;
	int i;
	int j;

	for (i = 0; i < subset_entries; i++) {
	for (j = 0; j < *fullset_entries; j++) {
	if (srp_addr_equal (&fullset[j], &subset[i])) {
	found = 1;
	break;
	}
	}
	if (found == 0) {
	srp_addr_copy (&fullset[*fullset_entries], &subset[i]);
	fullset_entries = fullset_entries + 1;
	}
	found = 0;
	}
	return;
	}

	static void memb_set_and_with_ring_id (
	struct srp_addr *set1,
	struct memb_ring_id *set1_ring_ids,
	int set1_entries,
	struct srp_addr *set2,
	int set2_entries,
	struct memb_ring_id *old_ring_id,
	struct srp_addr *and,
	int *and_entries)
	{
	int i;
	int j;
	int found = 0;

	*and_entries = 0;

	for (i = 0; i < set2_entries; i++) {
	for (j = 0; j < set1_entries; j++) {
	if (srp_addr_equal (&set1[j], &set2[i])) {
	if (memcmp (&set1_ring_ids[j], old_ring_id, sizeof (struct memb_ring_id)) == 0) {
	found = 1;
	}
	break;
	}
	}
	if (found) {
	srp_addr_copy (&and[*and_entries], &set1[j]);
	and_entries = and_entries + 1;
	}
	found = 0;
	}
	return;
	}

	#ifdef CODE_COVERAGE
	static void memb_set_print (
	char *string,
	struct srp_addr *list,
	int list_entries)
	{
	int i;
	int j;
	printf ("List '%s' contains %d entries:\n", string, list_entries);

	for (i = 0; i < list_entries; i++) {
	for (j = 0; j < INTERFACE_MAX; j++) {
	printf ("Address %d\n", i);
	printf ("\tiface %d %s\n", j, totemip_print (&list[i].addr[j]));
	printf ("family %d\n", list[i].addr[j].family);
	}
	}
	}
	#endif

	static void totemsrp_buffer_alloc (struct totemsrp_instance instance)
	{
	assert (instance != NULL);
	return totemrrp_buffer_alloc (instance->totemrrp_context);
	}

	static void totemsrp_buffer_release (struct totemsrp_instance instance, void ptr)
	{
	assert (instance != NULL);
	totemrrp_buffer_release (instance->totemrrp_context, ptr);
	}

	static void reset_token_retransmit_timeout (struct totemsrp_instance *instance)
	{
	qb_loop_timer_del (instance->totemsrp_poll_handle,
	instance->timer_orf_token_retransmit_timeout);
	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->token_retransmit_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	timer_function_token_retransmit_timeout,
	&instance->timer_orf_token_retransmit_timeout);

	}

	static void start_merge_detect_timeout (struct totemsrp_instance *instance)
	{
	if (instance->my_merge_detect_timeout_outstanding == 0) {
	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->merge_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	timer_function_merge_detect_timeout,
	&instance->timer_merge_detect_timeout);

	instance->my_merge_detect_timeout_outstanding = 1;
	}
	}

	static void cancel_merge_detect_timeout (struct totemsrp_instance *instance)
	{
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_merge_detect_timeout);
	instance->my_merge_detect_timeout_outstanding = 0;
	}

	/*
	* ring_state_* is used to save and restore the sort queue
	* state when a recovery operation fails (and enters gather)
	*/
	static void old_ring_state_save (struct totemsrp_instance *instance)
	{
	if (instance->old_ring_state_saved == 0) {
	instance->old_ring_state_saved = 1;
	memcpy (&instance->my_old_ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id));
	instance->old_ring_state_aru = instance->my_aru;
	instance->old_ring_state_high_seq_received = instance->my_high_seq_received;
	log_printf (instance->totemsrp_log_level_debug,
	"Saving state aru %x high seq received %x\n",
	instance->my_aru, instance->my_high_seq_received);
	}
	}

	static void old_ring_state_restore (struct totemsrp_instance *instance)
	{
	instance->my_aru = instance->old_ring_state_aru;
	instance->my_high_seq_received = instance->old_ring_state_high_seq_received;
	log_printf (instance->totemsrp_log_level_debug,
	"Restoring instance->my_aru %x my high seq received %x\n",
	instance->my_aru, instance->my_high_seq_received);
	}

	static void old_ring_state_reset (struct totemsrp_instance *instance)
	{
	log_printf (instance->totemsrp_log_level_debug,
	"Resetting old ring state\n");
	instance->old_ring_state_saved = 0;
	}

	static void reset_pause_timeout (struct totemsrp_instance *instance)
	{
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_pause_timeout);
	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->token_timeout * QB_TIME_NS_IN_MSEC / 5,
	(void *)instance,
	timer_function_pause_timeout,
	&instance->timer_pause_timeout);
	}

	static void reset_token_timeout (struct totemsrp_instance *instance) {
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout);
	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->token_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	timer_function_orf_token_timeout,
	&instance->timer_orf_token_timeout);
	}

	static void reset_heartbeat_timeout (struct totemsrp_instance *instance) {
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout);
	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->heartbeat_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	timer_function_heartbeat_timeout,
	&instance->timer_heartbeat_timeout);
	}


	static void cancel_token_timeout (struct totemsrp_instance *instance) {
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout);
	}

	static void cancel_heartbeat_timeout (struct totemsrp_instance *instance) {
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout);
	}

	static void cancel_token_retransmit_timeout (struct totemsrp_instance *instance)
	{
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_retransmit_timeout);
	}

	static void start_token_hold_retransmit_timeout (struct totemsrp_instance *instance)
	{
	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->token_hold_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	timer_function_token_hold_retransmit_timeout,
	&instance->timer_orf_token_hold_retransmit_timeout);
	}

	static void cancel_token_hold_retransmit_timeout (struct totemsrp_instance *instance)
	{
	qb_loop_timer_del (instance->totemsrp_poll_handle,
	instance->timer_orf_token_hold_retransmit_timeout);
	}

	static void memb_state_consensus_timeout_expired (
	struct totemsrp_instance *instance)
	{
	struct srp_addr no_consensus_list[PROCESSOR_COUNT_MAX];
	int no_consensus_list_entries;

	instance->stats.consensus_timeouts++;
	if (memb_consensus_agreed (instance)) {
	memb_consensus_reset (instance);

	memb_consensus_set (instance, &instance->my_id);

	reset_token_timeout (instance); // REVIEWED
	} else {
	memb_consensus_notset (
	instance,
	no_consensus_list,
	&no_consensus_list_entries,
	instance->my_proc_list,
	instance->my_proc_list_entries);

	memb_set_merge (no_consensus_list, no_consensus_list_entries,
	instance->my_failed_list, &instance->my_failed_list_entries);
	memb_state_gather_enter (instance, 0);
	}
	}

	static void memb_join_message_send (struct totemsrp_instance *instance);

	static void memb_merge_detect_transmit (struct totemsrp_instance *instance);

	/*
	* Timers used for various states of the membership algorithm
	*/
	static void timer_function_pause_timeout (void *data)
	{
	struct totemsrp_instance *instance = data;

	instance->pause_timestamp = qb_util_nano_current_get ();
	reset_pause_timeout (instance);
	}

	static void memb_recovery_state_token_loss (struct totemsrp_instance *instance)
	{
	old_ring_state_restore (instance);
	memb_state_gather_enter (instance, 5);
	instance->stats.recovery_token_lost++;
	}

	static void timer_function_orf_token_timeout (void *data)
	{
	struct totemsrp_instance *instance = data;

	switch (instance->memb_state) {
	case MEMB_STATE_OPERATIONAL:
	log_printf (instance->totemsrp_log_level_debug,
	"The token was lost in the OPERATIONAL state.\n");
	log_printf (instance->totemsrp_log_level_notice,
	"A processor failed, forming new configuration.\n");
	totemrrp_iface_check (instance->totemrrp_context);
	memb_state_gather_enter (instance, 2);
	instance->stats.operational_token_lost++;
	break;

	case MEMB_STATE_GATHER:
	log_printf (instance->totemsrp_log_level_debug,
	"The consensus timeout expired.\n");
	memb_state_consensus_timeout_expired (instance);
	memb_state_gather_enter (instance, 3);
	instance->stats.gather_token_lost++;
	break;

	case MEMB_STATE_COMMIT:
	log_printf (instance->totemsrp_log_level_debug,
	"The token was lost in the COMMIT state.\n");
	memb_state_gather_enter (instance, 4);
	instance->stats.commit_token_lost++;
	break;

	case MEMB_STATE_RECOVERY:
	log_printf (instance->totemsrp_log_level_debug,
	"The token was lost in the RECOVERY state.\n");
	memb_recovery_state_token_loss (instance);
	instance->orf_token_discard = 1;
	break;
	}
	}

	static void timer_function_heartbeat_timeout (void *data)
	{
	struct totemsrp_instance *instance = data;
	log_printf (instance->totemsrp_log_level_debug,
	"HeartBeat Timer expired Invoking token loss mechanism in state %d \n", instance->memb_state);
	timer_function_orf_token_timeout(data);
	}

	static void memb_timer_function_state_gather (void *data)
	{
	struct totemsrp_instance *instance = data;

	switch (instance->memb_state) {
	case MEMB_STATE_OPERATIONAL:
	case MEMB_STATE_RECOVERY:
	assert (0); /* this should never happen */
	break;
	case MEMB_STATE_GATHER:
	case MEMB_STATE_COMMIT:
	memb_join_message_send (instance);

	/*
	* Restart the join timeout
	`*/
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);

	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->join_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	memb_timer_function_state_gather,
	&instance->memb_timer_state_gather_join_timeout);
	break;
	}
	}

	static void memb_timer_function_gather_consensus_timeout (void *data)
	{
	struct totemsrp_instance *instance = data;
	memb_state_consensus_timeout_expired (instance);
	}

	static void deliver_messages_from_recovery_to_regular (struct totemsrp_instance *instance)
	{
	unsigned int i;
	struct sort_queue_item *recovery_message_item;
	struct sort_queue_item regular_message_item;
	unsigned int range = 0;
	int res;
	void *ptr;
	struct mcast *mcast;

	log_printf (instance->totemsrp_log_level_debug,
	"recovery to regular %x-%x\n", SEQNO_START_MSG + 1, instance->my_aru);

	range = instance->my_aru - SEQNO_START_MSG;
	/*
	* Move messages from recovery to regular sort queue
	*/
	// todo should i be initialized to 0 or 1 ?
	for (i = 1; i <= range; i++) {
	res = sq_item_get (&instance->recovery_sort_queue,
	i + SEQNO_START_MSG, &ptr);
	if (res != 0) {
	continue;
	}
	recovery_message_item = ptr;

	/*
	* Convert recovery message into regular message
	*/
	mcast = recovery_message_item->mcast;
	if (mcast->header.encapsulated == MESSAGE_ENCAPSULATED) {
	/*
	* Message is a recovery message encapsulated
	* in a new ring message
	*/
	regular_message_item.mcast =
	(struct mcast )(((char )recovery_message_item->mcast) + sizeof (struct mcast));
	regular_message_item.msg_len =
	recovery_message_item->msg_len - sizeof (struct mcast);
	mcast = regular_message_item.mcast;
	} else {
	/*
	* TODO this case shouldn't happen
	*/
	continue;
	}

	log_printf (instance->totemsrp_log_level_debug,
	"comparing if ring id is for this processors old ring seqno %d\n",
	mcast->seq);

	/*
	* Only add this message to the regular sort
	* queue if it was originated with the same ring
	* id as the previous ring
	*/
	if (memcmp (&instance->my_old_ring_id, &mcast->ring_id,
	sizeof (struct memb_ring_id)) == 0) {

	res = sq_item_inuse (&instance->regular_sort_queue, mcast->seq);
	if (res == 0) {
	sq_item_add (&instance->regular_sort_queue,
	&regular_message_item, mcast->seq);
	if (sq_lt_compare (instance->old_ring_state_high_seq_received, mcast->seq)) {
	instance->old_ring_state_high_seq_received = mcast->seq;
	}
	}
	} else {
	log_printf (instance->totemsrp_log_level_debug,
	"-not adding msg with seq no %x\n", mcast->seq);
	}
	}
	}

	/*
	* Change states in the state machine of the membership algorithm
	*/
	static void memb_state_operational_enter (struct totemsrp_instance *instance)
	{
	struct srp_addr joined_list[PROCESSOR_COUNT_MAX];
	int joined_list_entries = 0;
	unsigned int aru_save;
	unsigned int joined_list_totemip[PROCESSOR_COUNT_MAX];
	unsigned int trans_memb_list_totemip[PROCESSOR_COUNT_MAX];
	unsigned int new_memb_list_totemip[PROCESSOR_COUNT_MAX];
	unsigned int left_list[PROCESSOR_COUNT_MAX];
	unsigned int i;
	unsigned int res;

	memb_consensus_reset (instance);

	old_ring_state_reset (instance);

	deliver_messages_from_recovery_to_regular (instance);

	log_printf (instance->totemsrp_log_level_debug,
	"Delivering to app %x to %x\n",
	instance->my_high_delivered + 1, instance->old_ring_state_high_seq_received);

	aru_save = instance->my_aru;
	instance->my_aru = instance->old_ring_state_aru;

	messages_deliver_to_app (instance, 0, instance->old_ring_state_high_seq_received);

	/*
	* Calculate joined and left list
	*/
	memb_set_subtract (instance->my_left_memb_list,
	&instance->my_left_memb_entries,
	instance->my_memb_list, instance->my_memb_entries,
	instance->my_trans_memb_list, instance->my_trans_memb_entries);

	memb_set_subtract (joined_list, &joined_list_entries,
	instance->my_new_memb_list, instance->my_new_memb_entries,
	instance->my_trans_memb_list, instance->my_trans_memb_entries);

	/*
	* Install new membership
	*/
	instance->my_memb_entries = instance->my_new_memb_entries;
	memcpy (&instance->my_memb_list, instance->my_new_memb_list,
	sizeof (struct srp_addr) * instance->my_memb_entries);
	instance->last_released = 0;
	instance->my_set_retrans_flg = 0;

	/*
	* Deliver transitional configuration to application
	*/
	srp_addr_to_nodeid (left_list, instance->my_left_memb_list,
	instance->my_left_memb_entries);
	srp_addr_to_nodeid (trans_memb_list_totemip,
	instance->my_trans_memb_list, instance->my_trans_memb_entries);
	instance->totemsrp_confchg_fn (TOTEM_CONFIGURATION_TRANSITIONAL,
	trans_memb_list_totemip, instance->my_trans_memb_entries,
	left_list, instance->my_left_memb_entries,
	0, 0, &instance->my_ring_id);

	// TODO we need to filter to ensure we only deliver those
	// messages which are part of instance->my_deliver_memb
	messages_deliver_to_app (instance, 1, instance->old_ring_state_high_seq_received);

	instance->my_aru = aru_save;

	/*
	* Deliver regular configuration to application
	*/
	srp_addr_to_nodeid (new_memb_list_totemip,
	instance->my_new_memb_list, instance->my_new_memb_entries);
	srp_addr_to_nodeid (joined_list_totemip, joined_list,
	joined_list_entries);
	instance->totemsrp_confchg_fn (TOTEM_CONFIGURATION_REGULAR,
	new_memb_list_totemip, instance->my_new_memb_entries,
	0, 0,
	joined_list_totemip, joined_list_entries, &instance->my_ring_id);

	/*
	* The recovery sort queue now becomes the regular
	* sort queue. It is necessary to copy the state
	* into the regular sort queue.
	*/
	sq_copy (&instance->regular_sort_queue, &instance->recovery_sort_queue);
	instance->my_last_aru = SEQNO_START_MSG;

	/* When making my_proc_list smaller, ensure that the
	* now non-used entries are zero-ed out. There are some suspect
	* assert's that assume that there is always 2 entries in the list.
	* These fail when my_proc_list is reduced to 1 entry (and the
	* valid [0] entry is the same as the 'unused' [1] entry).
	*/
	memset(instance->my_proc_list, 0,
	sizeof (struct srp_addr) * instance->my_proc_list_entries);

	instance->my_proc_list_entries = instance->my_new_memb_entries;
	memcpy (instance->my_proc_list, instance->my_new_memb_list,
	sizeof (struct srp_addr) * instance->my_memb_entries);

	instance->my_failed_list_entries = 0;
	instance->my_high_delivered = instance->my_high_seq_received;

	for (i = 0; i <= instance->my_high_delivered; i++) {
	void *ptr;

	res = sq_item_get (&instance->regular_sort_queue, i, &ptr);
	if (res == 0) {
	struct sort_queue_item *regular_message;

	regular_message = ptr;
	free (regular_message->mcast);
	}
	}
	sq_items_release (&instance->regular_sort_queue, instance->my_high_delivered);
	instance->last_released = instance->my_high_delivered;

	log_printf (instance->totemsrp_log_level_debug,
	"entering OPERATIONAL state.\n");
	log_printf (instance->totemsrp_log_level_notice,
	"A processor joined or left the membership and a new membership was formed.\n");
	instance->memb_state = MEMB_STATE_OPERATIONAL;

	instance->stats.operational_entered++;
	instance->stats.continuous_gather = 0;

	instance->my_received_flg = 1;

	reset_pause_timeout (instance);

	/*
	* Save ring id information from this configuration to determine
	* which processors are transitioning from old regular configuration
	* in to new regular configuration on the next configuration change
	*/
	memcpy (&instance->my_old_ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id));

	return;
	}

	static void memb_state_gather_enter (
	struct totemsrp_instance *instance,
	int gather_from)
	{
	instance->orf_token_discard = 1;

	memb_set_merge (
	&instance->my_id, 1,
	instance->my_proc_list, &instance->my_proc_list_entries);

	memb_join_message_send (instance);

	/*
	* Restart the join timeout
	*/
	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);

	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->join_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	memb_timer_function_state_gather,
	&instance->memb_timer_state_gather_join_timeout);

	/*
	* Restart the consensus timeout
	*/
	qb_loop_timer_del (instance->totemsrp_poll_handle,
	instance->memb_timer_state_gather_consensus_timeout);

	qb_loop_timer_add (instance->totemsrp_poll_handle,
	QB_LOOP_MED,
	instance->totem_config->consensus_timeout*QB_TIME_NS_IN_MSEC,
	(void *)instance,
	memb_timer_function_gather_consensus_timeout,
	&instance->memb_timer_state_gather_consensus_timeout);

	/*
	* Cancel the token loss and token retransmission timeouts
	*/
	cancel_token_retransmit_timeout (instance); // REVIEWED
	cancel_token_timeout (instance); // REVIEWED
	cancel_merge_detect_timeout (instance);

	memb_consensus_reset (instance);

	memb_consensus_set (instance, &instance->my_id);

	log_printf (instance->totemsrp_log_level_debug,
	"entering GATHER state from %d.\n", gather_from);

	instance->memb_state = MEMB_STATE_GATHER;
	instance->stats.gather_entered++;

	if (gather_from == 3) {
	/*
	* State 3 means gather, so we are continuously gathering.
	*/
	instance->stats.continuous_gather++;
	}

	if (instance->stats.continuous_gather > MAX_NO_CONT_GATHER) {
	log_printf (instance->totemsrp_log_level_warning,
	"Totem is unable to form a cluster because of an "
	"operating system or network fault. The most common "
	"cause of this message is that the local firewall is "
	"configured improperly.\n");
	}

	return;
	}

	static void timer_function_token_retransmit_timeout (void *data);

	static void target_set_completed (
	void *context)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )context;

	memb_state_commit_token_send (instance);

	}

	static void memb_state_commit_enter (
	struct totemsrp_instance *instance)
	{
	old_ring_state_save (instance);

	memb_state_commit_token_update (instance);

	memb_state_commit_token_target_set (instance);

	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);

	instance->memb_timer_state_gather_join_timeout = 0;

	qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_consensus_timeout);

	instance->memb_timer_state_gather_consensus_timeout = 0;

	memb_ring_id_set_and_store (instance, &instance->commit_token->ring_id);

	instance->token_ring_id_seq = instance->my_ring_id.seq;

	log_printf (instance->totemsrp_log_level_debug,
	"entering COMMIT state.\n");

	instance->memb_state = MEMB_STATE_COMMIT;
	reset_token_retransmit_timeout (instance); // REVIEWED
	reset_token_timeout (instance); // REVIEWED

	instance->stats.commit_entered++;
	instance->stats.continuous_gather = 0;

	/*
	* reset all flow control variables since we are starting a new ring
	*/
	instance->my_trc = 0;
	instance->my_pbl = 0;
	instance->my_cbl = 0;
	/*
	* commit token sent after callback that token target has been set
	*/
	}

	static void memb_state_recovery_enter (
	struct totemsrp_instance *instance,
	struct memb_commit_token *commit_token)
	{
	int i;
	int local_received_flg = 1;
	unsigned int low_ring_aru;
	unsigned int range = 0;
	unsigned int messages_originated = 0;
	const struct srp_addr *addr;
	struct memb_commit_token_memb_entry *memb_list;
	struct memb_ring_id my_new_memb_ring_id_list[PROCESSOR_COUNT_MAX];

	addr = (const struct srp_addr *)commit_token->end_of_commit_token;
	memb_list = (struct memb_commit_token_memb_entry *)(addr + commit_token->addr_entries);

	log_printf (instance->totemsrp_log_level_debug,
	"entering RECOVERY state.\n");

	instance->orf_token_discard = 0;

	instance->my_high_ring_delivered = 0;

	sq_reinit (&instance->recovery_sort_queue, SEQNO_START_MSG);
	cs_queue_reinit (&instance->retrans_message_queue);

	low_ring_aru = instance->old_ring_state_high_seq_received;

	memb_state_commit_token_send_recovery (instance, commit_token);

	instance->my_token_seq = SEQNO_START_TOKEN - 1;

	/*
	* Build regular configuration
	*/
	totemrrp_processor_count_set (
	instance->totemrrp_context,
	commit_token->addr_entries);

	/*
	* Build transitional configuration
	*/
	for (i = 0; i < instance->my_new_memb_entries; i++) {
	memcpy (&my_new_memb_ring_id_list[i],
	&memb_list[i].ring_id,
	sizeof (struct memb_ring_id));
	}
	memb_set_and_with_ring_id (
	instance->my_new_memb_list,
	my_new_memb_ring_id_list,
	instance->my_new_memb_entries,
	instance->my_memb_list,
	instance->my_memb_entries,
	&instance->my_old_ring_id,
	instance->my_trans_memb_list,
	&instance->my_trans_memb_entries);

	for (i = 0; i < instance->my_trans_memb_entries; i++) {
	log_printf (instance->totemsrp_log_level_debug,
	"TRANS [%d] member %s:\n", i, totemip_print (&instance->my_trans_memb_list[i].addr[0]));
	}
	for (i = 0; i < instance->my_new_memb_entries; i++) {
	log_printf (instance->totemsrp_log_level_debug,
	"position [%d] member %s:\n", i, totemip_print (&addr[i].addr[0]));
	log_printf (instance->totemsrp_log_level_debug,
	"previous ring seq %lld rep %s\n",
	memb_list[i].ring_id.seq,
	totemip_print (&memb_list[i].ring_id.rep));

	log_printf (instance->totemsrp_log_level_debug,
	"aru %x high delivered %x received flag %d\n",
	memb_list[i].aru,
	memb_list[i].high_delivered,
	memb_list[i].received_flg);

	// assert (totemip_print (&memb_list[i].ring_id.rep) != 0);
	}
	/*
	* Determine if any received flag is false
	*/
	for (i = 0; i < commit_token->addr_entries; i++) {
	if (memb_set_subset (&instance->my_new_memb_list[i], 1,
	instance->my_trans_memb_list, instance->my_trans_memb_entries) &&

	memb_list[i].received_flg == 0) {
	instance->my_deliver_memb_entries = instance->my_trans_memb_entries;
	memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list,
	sizeof (struct srp_addr) * instance->my_trans_memb_entries);
	local_received_flg = 0;
	break;
	}
	}
	if (local_received_flg == 1) {
	goto no_originate;
	} /* Else originate messages if we should */

	/*
	* Calculate my_low_ring_aru, instance->my_high_ring_delivered for the transitional membership
	*/
	for (i = 0; i < commit_token->addr_entries; i++) {
	if (memb_set_subset (&instance->my_new_memb_list[i], 1,
	instance->my_deliver_memb_list,
	instance->my_deliver_memb_entries) &&

	memcmp (&instance->my_old_ring_id,
	&memb_list[i].ring_id,
	sizeof (struct memb_ring_id)) == 0) {

	if (sq_lt_compare (memb_list[i].aru, low_ring_aru)) {

	low_ring_aru = memb_list[i].aru;
	}
	if (sq_lt_compare (instance->my_high_ring_delivered, memb_list[i].high_delivered)) {
	instance->my_high_ring_delivered = memb_list[i].high_delivered;
	}
	}
	}

	/*
	* Copy all old ring messages to instance->retrans_message_queue
	*/
	range = instance->old_ring_state_high_seq_received - low_ring_aru;
	if (range == 0) {
	/*
	* No messages to copy
	*/
	goto no_originate;
	}
	assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);

	log_printf (instance->totemsrp_log_level_debug,
	"copying all old ring messages from %x-%x.\n",
	low_ring_aru + 1, instance->old_ring_state_high_seq_received);

	for (i = 1; i <= range; i++) {
	struct sort_queue_item *sort_queue_item;
	struct message_item message_item;
	void *ptr;
	int res;

	res = sq_item_get (&instance->regular_sort_queue,
	low_ring_aru + i, &ptr);
	if (res != 0) {
	continue;
	}
	sort_queue_item = ptr;
	messages_originated++;
	memset (&message_item, 0, sizeof (struct message_item));
	// TODO LEAK
	message_item.mcast = totemsrp_buffer_alloc (instance);
	assert (message_item.mcast);
	message_item.mcast->header.type = MESSAGE_TYPE_MCAST;
	srp_addr_copy (&message_item.mcast->system_from, &instance->my_id);
	message_item.mcast->header.encapsulated = MESSAGE_ENCAPSULATED;
	message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid;
	assert (message_item.mcast->header.nodeid);
	message_item.mcast->header.endian_detector = ENDIAN_LOCAL;
	memcpy (&message_item.mcast->ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id));
	message_item.msg_len = sort_queue_item->msg_len + sizeof (struct mcast);
	memcpy (((char *)message_item.mcast) + sizeof (struct mcast),
	sort_queue_item->mcast,
	sort_queue_item->msg_len);
	cs_queue_item_add (&instance->retrans_message_queue, &message_item);
	}
	log_printf (instance->totemsrp_log_level_debug,
	"Originated %d messages in RECOVERY.\n", messages_originated);
	goto originated;

	no_originate:
	log_printf (instance->totemsrp_log_level_debug,
	"Did not need to originate any messages in recovery.\n");

	originated:
	instance->my_aru = SEQNO_START_MSG;
	instance->my_aru_count = 0;
	instance->my_seq_unchanged = 0;
	instance->my_high_seq_received = SEQNO_START_MSG;
	instance->my_install_seq = SEQNO_START_MSG;
	instance->last_released = SEQNO_START_MSG;

	reset_token_timeout (instance); // REVIEWED
	reset_token_retransmit_timeout (instance); // REVIEWED

	instance->memb_state = MEMB_STATE_RECOVERY;
	instance->stats.recovery_entered++;
	instance->stats.continuous_gather = 0;

	return;
	}

	void totemsrp_event_signal (void *srp_context, enum totem_event_type type, int value)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;

	token_hold_cancel_send (instance);

	return;
	}

	int totemsrp_mcast (
	void *srp_context,
	struct iovec *iovec,
	unsigned int iov_len,
	int guarantee)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;
	int i;
	struct message_item message_item;
	char *addr;
	unsigned int addr_idx;

	if (cs_queue_is_full (&instance->new_message_queue)) {
	log_printf (instance->totemsrp_log_level_debug, "queue full\n");
	return (-1);
	}

	memset (&message_item, 0, sizeof (struct message_item));

	/*
	* Allocate pending item
	*/
	message_item.mcast = totemsrp_buffer_alloc (instance);
	if (message_item.mcast == 0) {
	goto error_mcast;
	}

	/*
	* Set mcast header
	*/
	memset(message_item.mcast, 0, sizeof (struct mcast));
	message_item.mcast->header.type = MESSAGE_TYPE_MCAST;
	message_item.mcast->header.endian_detector = ENDIAN_LOCAL;
	message_item.mcast->header.encapsulated = MESSAGE_NOT_ENCAPSULATED;
	message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid;
	assert (message_item.mcast->header.nodeid);

	message_item.mcast->guarantee = guarantee;
	srp_addr_copy (&message_item.mcast->system_from, &instance->my_id);

	addr = (char *)message_item.mcast;
	addr_idx = sizeof (struct mcast);
	for (i = 0; i < iov_len; i++) {
	memcpy (&addr[addr_idx], iovec[i].iov_base, iovec[i].iov_len);
	addr_idx += iovec[i].iov_len;
	}

	message_item.msg_len = addr_idx;

	log_printf (instance->totemsrp_log_level_debug, "mcasted message added to pending queue\n");
	instance->stats.mcast_tx++;
	cs_queue_item_add (&instance->new_message_queue, &message_item);

	return (0);

	error_mcast:
	return (-1);
	}

	/*
	* Determine if there is room to queue a new message
	*/
	int totemsrp_avail (void *srp_context)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;
	int avail;

	cs_queue_avail (&instance->new_message_queue, &avail);

	return (avail);
	}

	/*
	* ORF Token Management
	*/
	/*
	* Recast message to mcast group if it is available
	*/
	static int orf_token_remcast (
	struct totemsrp_instance *instance,
	int seq)
	{
	struct sort_queue_item *sort_queue_item;
	int res;
	void *ptr;

	struct sq *sort_queue;

	if (instance->memb_state == MEMB_STATE_RECOVERY) {
	sort_queue = &instance->recovery_sort_queue;
	} else {
	sort_queue = &instance->regular_sort_queue;
	}

	res = sq_in_range (sort_queue, seq);
	if (res == 0) {
	log_printf (instance->totemsrp_log_level_debug, "sq not in range\n");
	return (-1);
	}

	/*
	* Get RTR item at seq, if not available, return
	*/
	res = sq_item_get (sort_queue, seq, &ptr);
	if (res != 0) {
	return -1;
	}

	sort_queue_item = ptr;

	totemrrp_mcast_noflush_send (
	instance->totemrrp_context,
	sort_queue_item->mcast,
	sort_queue_item->msg_len);

	return (0);
	}


	/*
	* Free all freeable messages from ring
	*/
	static void messages_free (
	struct totemsrp_instance *instance,
	unsigned int token_aru)
	{
	struct sort_queue_item *regular_message;
	unsigned int i;
	int res;
	int log_release = 0;
	unsigned int release_to;
	unsigned int range = 0;

	release_to = token_aru;
	if (sq_lt_compare (instance->my_last_aru, release_to)) {
	release_to = instance->my_last_aru;
	}
	if (sq_lt_compare (instance->my_high_delivered, release_to)) {
	release_to = instance->my_high_delivered;
	}

	/*
	* Ensure we dont try release before an already released point
	*/
	if (sq_lt_compare (release_to, instance->last_released)) {
	return;
	}

	range = release_to - instance->last_released;
	assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);

	/*
	* Release retransmit list items if group aru indicates they are transmitted
	*/
	for (i = 1; i <= range; i++) {
	void *ptr;

	res = sq_item_get (&instance->regular_sort_queue,
	instance->last_released + i, &ptr);
	if (res == 0) {
	regular_message = ptr;
	totemsrp_buffer_release (instance, regular_message->mcast);
	}
	sq_items_release (&instance->regular_sort_queue,
	instance->last_released + i);

	log_release = 1;
	}
	instance->last_released += range;

	if (log_release) {
	log_printf (instance->totemsrp_log_level_debug,
	"releasing messages up to and including %x\n", release_to);
	}
	}

	static void update_aru (
	struct totemsrp_instance *instance)
	{
	unsigned int i;
	int res;
	struct sq *sort_queue;
	unsigned int range;
	unsigned int my_aru_saved = 0;

	if (instance->memb_state == MEMB_STATE_RECOVERY) {
	sort_queue = &instance->recovery_sort_queue;
	} else {
	sort_queue = &instance->regular_sort_queue;
	}

	range = instance->my_high_seq_received - instance->my_aru;
	if (range > 1024) {
	return;
	}

	my_aru_saved = instance->my_aru;
	for (i = 1; i <= range; i++) {

	void *ptr;

	res = sq_item_get (sort_queue, my_aru_saved + i, &ptr);
	/*
	* If hole, stop updating aru
	*/
	if (res != 0) {
	break;
	}
	}
	instance->my_aru += i - 1;
	}

	/*
	* Multicasts pending messages onto the ring (requires orf_token possession)
	*/
	static int orf_token_mcast (
	struct totemsrp_instance *instance,
	struct orf_token *token,
	int fcc_mcasts_allowed)
	{
	struct message_item *message_item = 0;
	struct cs_queue *mcast_queue;
	struct sq *sort_queue;
	struct sort_queue_item sort_queue_item;
	struct mcast *mcast;
	unsigned int fcc_mcast_current;

	if (instance->memb_state == MEMB_STATE_RECOVERY) {
	mcast_queue = &instance->retrans_message_queue;
	sort_queue = &instance->recovery_sort_queue;
	reset_token_retransmit_timeout (instance); // REVIEWED
	} else {
	mcast_queue = &instance->new_message_queue;
	sort_queue = &instance->regular_sort_queue;
	}

	for (fcc_mcast_current = 0; fcc_mcast_current < fcc_mcasts_allowed; fcc_mcast_current++) {
	if (cs_queue_is_empty (mcast_queue)) {
	break;
	}
	message_item = (struct message_item *)cs_queue_item_get (mcast_queue);

	message_item->mcast->seq = ++token->seq;
	message_item->mcast->this_seqno = instance->global_seqno++;

	/*
	* Build IO vector
	*/
	memset (&sort_queue_item, 0, sizeof (struct sort_queue_item));
	sort_queue_item.mcast = message_item->mcast;
	sort_queue_item.msg_len = message_item->msg_len;

	mcast = sort_queue_item.mcast;

	memcpy (&mcast->ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id));

	/*
	* Add message to retransmit queue
	*/
	sq_item_add (sort_queue, &sort_queue_item, message_item->mcast->seq);

	totemrrp_mcast_noflush_send (
	instance->totemrrp_context,
	message_item->mcast,
	message_item->msg_len);

	/*
	* Delete item from pending queue
	*/
	cs_queue_item_remove (mcast_queue);

	/*
	* If messages mcasted, deliver any new messages to totempg
	*/
	instance->my_high_seq_received = token->seq;
	}

	update_aru (instance);

	/*
	* Return 1 if more messages are available for single node clusters
	*/
	return (fcc_mcast_current);
	}

	/*
	* Remulticasts messages in orf_token's retransmit list (requires orf_token)
	* Modify's orf_token's rtr to include retransmits required by this process
	*/
	static int orf_token_rtr (
	struct totemsrp_instance *instance,
	struct orf_token *orf_token,
	unsigned int *fcc_allowed)
	{
	unsigned int res;
	unsigned int i, j;
	unsigned int found;
	struct sq *sort_queue;
	struct rtr_item *rtr_list;
	unsigned int range = 0;
	char retransmit_msg[1024];
	char value[64];

	if (instance->memb_state == MEMB_STATE_RECOVERY) {
	sort_queue = &instance->recovery_sort_queue;
	} else {
	sort_queue = &instance->regular_sort_queue;
	}

	rtr_list = &orf_token->rtr_list[0];

	strcpy (retransmit_msg, "Retransmit List: ");
	if (orf_token->rtr_list_entries) {
	log_printf (instance->totemsrp_log_level_debug,
	"Retransmit List %d\n", orf_token->rtr_list_entries);
	for (i = 0; i < orf_token->rtr_list_entries; i++) {
	sprintf (value, "%x ", rtr_list[i].seq);
	strcat (retransmit_msg, value);
	}
	strcat (retransmit_msg, "\n");
	log_printf (instance->totemsrp_log_level_notice,
	"%s", retransmit_msg);
	}

	/*
	* Retransmit messages on orf_token's RTR list from RTR queue
	*/
	for (instance->fcc_remcast_current = 0, i = 0;
	instance->fcc_remcast_current < *fcc_allowed && i < orf_token->rtr_list_entries;) {

	/*
	* If this retransmit request isn't from this configuration,
	* try next rtr entry
	*/
	if (memcmp (&rtr_list[i].ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id)) != 0) {

	i += 1;
	continue;
	}

	res = orf_token_remcast (instance, rtr_list[i].seq);
	if (res == 0) {
	/*
	* Multicasted message, so no need to copy to new retransmit list
	*/
	orf_token->rtr_list_entries -= 1;
	assert (orf_token->rtr_list_entries >= 0);
	memmove (&rtr_list[i], &rtr_list[i + 1],
	sizeof (struct rtr_item) * (orf_token->rtr_list_entries - i));

	instance->stats.mcast_retx++;
	instance->fcc_remcast_current++;
	} else {
	i += 1;
	}
	}
	fcc_allowed = fcc_allowed - instance->fcc_remcast_current;

	/*
	* Add messages to retransmit to RTR list
	* but only retry if there is room in the retransmit list
	*/

	range = orf_token->seq - instance->my_aru;
	assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);

	for (i = 1; (orf_token->rtr_list_entries < RETRANSMIT_ENTRIES_MAX) &&
	(i <= range); i++) {

	/*
	* Ensure message is within the sort queue range
	*/
	res = sq_in_range (sort_queue, instance->my_aru + i);
	if (res == 0) {
	break;
	}

	/*
	* Find if a message is missing from this processor
	*/
	res = sq_item_inuse (sort_queue, instance->my_aru + i);
	if (res == 0) {
	/*
	* Determine how many times we have missed receiving
	* this sequence number. sq_item_miss_count increments
	* a counter for the sequence number. The miss count
	* will be returned and compared. This allows time for
	* delayed multicast messages to be received before
	* declaring the message is missing and requesting a
	* retransmit.
	*/
	res = sq_item_miss_count (sort_queue, instance->my_aru + i);
	if (res < instance->totem_config->miss_count_const) {
	continue;
	}

	/*
	* Determine if missing message is already in retransmit list
	*/
	found = 0;
	for (j = 0; j < orf_token->rtr_list_entries; j++) {
	if (instance->my_aru + i == rtr_list[j].seq) {
	found = 1;
	}
	}
	if (found == 0) {
	/*
	* Missing message not found in current retransmit list so add it
	*/
	memcpy (&rtr_list[orf_token->rtr_list_entries].ring_id,
	&instance->my_ring_id, sizeof (struct memb_ring_id));
	rtr_list[orf_token->rtr_list_entries].seq = instance->my_aru + i;
	orf_token->rtr_list_entries++;
	}
	}
	}
	return (instance->fcc_remcast_current);
	}

	static void token_retransmit (struct totemsrp_instance *instance)
	{
	totemrrp_token_send (instance->totemrrp_context,
	instance->orf_token_retransmit,
	instance->orf_token_retransmit_size);
	}

	/*
	* Retransmit the regular token if no mcast or token has
	* been received in retransmit token period retransmit
	* the token to the next processor
	*/
	static void timer_function_token_retransmit_timeout (void *data)
	{
	struct totemsrp_instance *instance = data;

	switch (instance->memb_state) {
	case MEMB_STATE_GATHER:
	break;
	case MEMB_STATE_COMMIT:
	case MEMB_STATE_OPERATIONAL:
	case MEMB_STATE_RECOVERY:
	token_retransmit (instance);
	reset_token_retransmit_timeout (instance); // REVIEWED
	break;
	}
	}

	static void timer_function_token_hold_retransmit_timeout (void *data)
	{
	struct totemsrp_instance *instance = data;

	switch (instance->memb_state) {
	case MEMB_STATE_GATHER:
	break;
	case MEMB_STATE_COMMIT:
	break;
	case MEMB_STATE_OPERATIONAL:
	case MEMB_STATE_RECOVERY:
	token_retransmit (instance);
	break;
	}
	}

	static void timer_function_merge_detect_timeout(void *data)
	{
	struct totemsrp_instance *instance = data;

	instance->my_merge_detect_timeout_outstanding = 0;

	switch (instance->memb_state) {
	case MEMB_STATE_OPERATIONAL:
	if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
	memb_merge_detect_transmit (instance);
	}
	break;
	case MEMB_STATE_GATHER:
	case MEMB_STATE_COMMIT:
	case MEMB_STATE_RECOVERY:
	break;
	}
	}

	/*
	* Send orf_token to next member (requires orf_token)
	*/
	static int token_send (
	struct totemsrp_instance *instance,
	struct orf_token *orf_token,
	int forward_token)
	{
	int res = 0;
	unsigned int orf_token_size;

	orf_token_size = sizeof (struct orf_token) +
	(orf_token->rtr_list_entries * sizeof (struct rtr_item));

	memcpy (instance->orf_token_retransmit, orf_token, orf_token_size);
	instance->orf_token_retransmit_size = orf_token_size;
	orf_token->header.nodeid = instance->my_id.addr[0].nodeid;
	assert (orf_token->header.nodeid);

	if (forward_token == 0) {
	return (0);
	}

	totemrrp_token_send (instance->totemrrp_context,
	orf_token,
	orf_token_size);

	return (res);
	}

	static int token_hold_cancel_send (struct totemsrp_instance *instance)
	{
	struct token_hold_cancel token_hold_cancel;

	/*
	* Only cancel if the token is currently held
	*/
	if (instance->my_token_held == 0) {
	return (0);
	}
	instance->my_token_held = 0;

	/*
	* Build message
	*/
	token_hold_cancel.header.type = MESSAGE_TYPE_TOKEN_HOLD_CANCEL;
	token_hold_cancel.header.endian_detector = ENDIAN_LOCAL;
	token_hold_cancel.header.encapsulated = 0;
	token_hold_cancel.header.nodeid = instance->my_id.addr[0].nodeid;
	memcpy (&token_hold_cancel.ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id));
	assert (token_hold_cancel.header.nodeid);

	instance->stats.token_hold_cancel_tx++;

	totemrrp_mcast_flush_send (instance->totemrrp_context, &token_hold_cancel,
	sizeof (struct token_hold_cancel));

	return (0);
	}

	static int orf_token_send_initial (struct totemsrp_instance *instance)
	{
	struct orf_token orf_token;
	int res;

	orf_token.header.type = MESSAGE_TYPE_ORF_TOKEN;
	orf_token.header.endian_detector = ENDIAN_LOCAL;
	orf_token.header.encapsulated = 0;
	orf_token.header.nodeid = instance->my_id.addr[0].nodeid;
	assert (orf_token.header.nodeid);
	orf_token.seq = SEQNO_START_MSG;
	orf_token.token_seq = SEQNO_START_TOKEN;
	orf_token.retrans_flg = 1;
	instance->my_set_retrans_flg = 1;
	instance->stats.orf_token_tx++;

	if (cs_queue_is_empty (&instance->retrans_message_queue) == 1) {
	orf_token.retrans_flg = 0;
	instance->my_set_retrans_flg = 0;
	} else {
	orf_token.retrans_flg = 1;
	instance->my_set_retrans_flg = 1;
	}

	orf_token.aru = 0;
	orf_token.aru = SEQNO_START_MSG - 1;
	orf_token.aru_addr = instance->my_id.addr[0].nodeid;

	memcpy (&orf_token.ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id));
	orf_token.fcc = 0;
	orf_token.backlog = 0;

	orf_token.rtr_list_entries = 0;

	res = token_send (instance, &orf_token, 1);

	return (res);
	}

	static void memb_state_commit_token_update (
	struct totemsrp_instance *instance)
	{
	struct srp_addr *addr;
	struct memb_commit_token_memb_entry *memb_list;
	unsigned int high_aru;
	unsigned int i;

	addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
	memb_list = (struct memb_commit_token_memb_entry *)(addr + instance->commit_token->addr_entries);

	memcpy (instance->my_new_memb_list, addr,
	sizeof (struct srp_addr) * instance->commit_token->addr_entries);

	instance->my_new_memb_entries = instance->commit_token->addr_entries;

	memcpy (&memb_list[instance->commit_token->memb_index].ring_id,
	&instance->my_old_ring_id, sizeof (struct memb_ring_id));

	memb_list[instance->commit_token->memb_index].aru = instance->old_ring_state_aru;
	/*
	* TODO high delivered is really instance->my_aru, but with safe this
	* could change?
	*/
	instance->my_received_flg =
	(instance->my_aru == instance->my_high_seq_received);

	memb_list[instance->commit_token->memb_index].received_flg = instance->my_received_flg;

	memb_list[instance->commit_token->memb_index].high_delivered = instance->my_high_delivered;
	/*
	* find high aru up to current memb_index for all matching ring ids
	* if any ring id matching memb_index has aru less then high aru set
	* received flag for that entry to false
	*/
	high_aru = memb_list[instance->commit_token->memb_index].aru;
	for (i = 0; i <= instance->commit_token->memb_index; i++) {
	if (memcmp (&memb_list[instance->commit_token->memb_index].ring_id,
	&memb_list[i].ring_id,
	sizeof (struct memb_ring_id)) == 0) {

	if (sq_lt_compare (high_aru, memb_list[i].aru)) {
	high_aru = memb_list[i].aru;
	}
	}
	}

	for (i = 0; i <= instance->commit_token->memb_index; i++) {
	if (memcmp (&memb_list[instance->commit_token->memb_index].ring_id,
	&memb_list[i].ring_id,
	sizeof (struct memb_ring_id)) == 0) {

	if (sq_lt_compare (memb_list[i].aru, high_aru)) {
	memb_list[i].received_flg = 0;
	if (i == instance->commit_token->memb_index) {
	instance->my_received_flg = 0;
	}
	}
	}
	}

	instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
	instance->commit_token->memb_index += 1;
	assert (instance->commit_token->memb_index <= instance->commit_token->addr_entries);
	assert (instance->commit_token->header.nodeid);
	}

	static void memb_state_commit_token_target_set (
	struct totemsrp_instance *instance)
	{
	struct srp_addr *addr;
	unsigned int i;

	addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;

	for (i = 0; i < instance->totem_config->interface_count; i++) {
	totemrrp_token_target_set (
	instance->totemrrp_context,
	&addr[instance->commit_token->memb_index %
	instance->commit_token->addr_entries].addr[i],
	i);
	}
	}

	static int memb_state_commit_token_send_recovery (
	struct totemsrp_instance *instance,
	struct memb_commit_token *commit_token)
	{
	unsigned int commit_token_size;

	commit_token->token_seq++;
	commit_token_size = sizeof (struct memb_commit_token) +
	((sizeof (struct srp_addr) +
	sizeof (struct memb_commit_token_memb_entry)) * commit_token->addr_entries);
	/*
	* Make a copy for retransmission if necessary
	*/
	memcpy (instance->orf_token_retransmit, commit_token, commit_token_size);
	instance->orf_token_retransmit_size = commit_token_size;

	instance->stats.memb_commit_token_tx++;

	totemrrp_token_send (instance->totemrrp_context,
	commit_token,
	commit_token_size);

	/*
	* Request retransmission of the commit token in case it is lost
	*/
	reset_token_retransmit_timeout (instance);
	return (0);
	}

	static int memb_state_commit_token_send (
	struct totemsrp_instance *instance)
	{
	unsigned int commit_token_size;

	instance->commit_token->token_seq++;
	commit_token_size = sizeof (struct memb_commit_token) +
	((sizeof (struct srp_addr) +
	sizeof (struct memb_commit_token_memb_entry)) * instance->commit_token->addr_entries);
	/*
	* Make a copy for retransmission if necessary
	*/
	memcpy (instance->orf_token_retransmit, instance->commit_token, commit_token_size);
	instance->orf_token_retransmit_size = commit_token_size;

	instance->stats.memb_commit_token_tx++;

	totemrrp_token_send (instance->totemrrp_context,
	instance->commit_token,
	commit_token_size);

	/*
	* Request retransmission of the commit token in case it is lost
	*/
	reset_token_retransmit_timeout (instance);
	return (0);
	}


	static int memb_lowest_in_config (struct totemsrp_instance *instance)
	{
	struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
	int token_memb_entries = 0;
	int i;
	struct totem_ip_address *lowest_addr;

	memb_set_subtract (token_memb, &token_memb_entries,
	instance->my_proc_list, instance->my_proc_list_entries,
	instance->my_failed_list, instance->my_failed_list_entries);

	/*
	* find representative by searching for smallest identifier
	*/

	lowest_addr = &token_memb[0].addr[0];
	for (i = 1; i < token_memb_entries; i++) {
	if (totemip_compare(lowest_addr, &token_memb[i].addr[0]) > 0) {
	totemip_copy (lowest_addr, &token_memb[i].addr[0]);
	}
	}
	return (totemip_compare (lowest_addr, &instance->my_id.addr[0]) == 0);
	}

	static int srp_addr_compare (const void a, const void b)
	{
	const struct srp_addr srp_a = (const struct srp_addr )a;
	const struct srp_addr srp_b = (const struct srp_addr )b;

	return (totemip_compare (&srp_a->addr[0], &srp_b->addr[0]));
	}

	static void memb_state_commit_token_create (
	struct totemsrp_instance *instance)
	{
	struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
	struct srp_addr *addr;
	struct memb_commit_token_memb_entry *memb_list;
	int token_memb_entries = 0;

	log_printf (instance->totemsrp_log_level_debug,
	"Creating commit token because I am the rep.\n");

	memb_set_subtract (token_memb, &token_memb_entries,
	instance->my_proc_list, instance->my_proc_list_entries,
	instance->my_failed_list, instance->my_failed_list_entries);

	memset (instance->commit_token, 0, sizeof (struct memb_commit_token));
	instance->commit_token->header.type = MESSAGE_TYPE_MEMB_COMMIT_TOKEN;
	instance->commit_token->header.endian_detector = ENDIAN_LOCAL;
	instance->commit_token->header.encapsulated = 0;
	instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
	assert (instance->commit_token->header.nodeid);

	totemip_copy(&instance->commit_token->ring_id.rep, &instance->my_id.addr[0]);

	instance->commit_token->ring_id.seq = instance->token_ring_id_seq + 4;

	/*
	* This qsort is necessary to ensure the commit token traverses
	* the ring in the proper order
	*/
	qsort (token_memb, token_memb_entries, sizeof (struct srp_addr),
	srp_addr_compare);

	instance->commit_token->memb_index = 0;
	instance->commit_token->addr_entries = token_memb_entries;

	addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
	memb_list = (struct memb_commit_token_memb_entry *)(addr + instance->commit_token->addr_entries);

	memcpy (addr, token_memb,
	token_memb_entries * sizeof (struct srp_addr));
	memset (memb_list, 0,
	sizeof (struct memb_commit_token_memb_entry) * token_memb_entries);
	}

	static void memb_join_message_send (struct totemsrp_instance *instance)
	{
	char memb_join_data[40000];
	struct memb_join memb_join = (struct memb_join )memb_join_data;
	char *addr;
	unsigned int addr_idx;

	memb_join->header.type = MESSAGE_TYPE_MEMB_JOIN;
	memb_join->header.endian_detector = ENDIAN_LOCAL;
	memb_join->header.encapsulated = 0;
	memb_join->header.nodeid = instance->my_id.addr[0].nodeid;
	assert (memb_join->header.nodeid);

	memb_join->ring_seq = instance->my_ring_id.seq;
	memb_join->proc_list_entries = instance->my_proc_list_entries;
	memb_join->failed_list_entries = instance->my_failed_list_entries;
	srp_addr_copy (&memb_join->system_from, &instance->my_id);

	/*
	* This mess adds the joined and failed processor lists into the join
	* message
	*/
	addr = (char *)memb_join;
	addr_idx = sizeof (struct memb_join);
	memcpy (&addr[addr_idx],
	instance->my_proc_list,
	instance->my_proc_list_entries *
	sizeof (struct srp_addr));
	addr_idx +=
	instance->my_proc_list_entries *
	sizeof (struct srp_addr);
	memcpy (&addr[addr_idx],
	instance->my_failed_list,
	instance->my_failed_list_entries *
	sizeof (struct srp_addr));
	addr_idx +=
	instance->my_failed_list_entries *
	sizeof (struct srp_addr);


	if (instance->totem_config->send_join_timeout) {
	usleep (random() % (instance->totem_config->send_join_timeout * 1000));
	}

	instance->stats.memb_join_tx++;

	totemrrp_mcast_flush_send (
	instance->totemrrp_context,
	memb_join,
	addr_idx);
	}

	static void memb_leave_message_send (struct totemsrp_instance *instance)
	{
	char memb_join_data[40000];
	struct memb_join memb_join = (struct memb_join )memb_join_data;
	char *addr;
	unsigned int addr_idx;
	int active_memb_entries;
	struct srp_addr active_memb[PROCESSOR_COUNT_MAX];

	log_printf (instance->totemsrp_log_level_debug,
	"sending join/leave message\n");

	/*
	* add us to the failed list, and remove us from
	* the members list
	*/
	memb_set_merge(
	&instance->my_id, 1,
	instance->my_failed_list, &instance->my_failed_list_entries);

	memb_set_subtract (active_memb, &active_memb_entries,
	instance->my_proc_list, instance->my_proc_list_entries,
	&instance->my_id, 1);


	memb_join->header.type = MESSAGE_TYPE_MEMB_JOIN;
	memb_join->header.endian_detector = ENDIAN_LOCAL;
	memb_join->header.encapsulated = 0;
	memb_join->header.nodeid = LEAVE_DUMMY_NODEID;

	memb_join->ring_seq = instance->my_ring_id.seq;
	memb_join->proc_list_entries = active_memb_entries;
	memb_join->failed_list_entries = instance->my_failed_list_entries;
	srp_addr_copy (&memb_join->system_from, &instance->my_id);
	memb_join->system_from.addr[0].nodeid = LEAVE_DUMMY_NODEID;

	// TODO: CC Maybe use the actual join send routine.
	/*
	* This mess adds the joined and failed processor lists into the join
	* message
	*/
	addr = (char *)memb_join;
	addr_idx = sizeof (struct memb_join);
	memcpy (&addr[addr_idx],
	active_memb,
	active_memb_entries *
	sizeof (struct srp_addr));
	addr_idx +=
	active_memb_entries *
	sizeof (struct srp_addr);
	memcpy (&addr[addr_idx],
	instance->my_failed_list,
	instance->my_failed_list_entries *
	sizeof (struct srp_addr));
	addr_idx +=
	instance->my_failed_list_entries *
	sizeof (struct srp_addr);


	if (instance->totem_config->send_join_timeout) {
	usleep (random() % (instance->totem_config->send_join_timeout * 1000));
	}
	instance->stats.memb_join_tx++;

	totemrrp_mcast_flush_send (
	instance->totemrrp_context,
	memb_join,
	addr_idx);
	}

	static void memb_merge_detect_transmit (struct totemsrp_instance *instance)
	{
	struct memb_merge_detect memb_merge_detect;

	memb_merge_detect.header.type = MESSAGE_TYPE_MEMB_MERGE_DETECT;
	memb_merge_detect.header.endian_detector = ENDIAN_LOCAL;
	memb_merge_detect.header.encapsulated = 0;
	memb_merge_detect.header.nodeid = instance->my_id.addr[0].nodeid;
	srp_addr_copy (&memb_merge_detect.system_from, &instance->my_id);
	memcpy (&memb_merge_detect.ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id));
	assert (memb_merge_detect.header.nodeid);

	instance->stats.memb_merge_detect_tx++;
	totemrrp_mcast_flush_send (instance->totemrrp_context,
	&memb_merge_detect,
	sizeof (struct memb_merge_detect));
	}

	static void memb_ring_id_create_or_load (
	struct totemsrp_instance *instance,
	struct memb_ring_id *memb_ring_id)
	{
	int fd;
	int res = 0;
	char filename[PATH_MAX];

	snprintf (filename, sizeof(filename), "%s/ringid_%s",
	rundir, totemip_print (&instance->my_id.addr[0]));
	fd = open (filename, O_RDONLY, 0700);
	/*
	* If file can be opened and read, read the ring id
	*/
	if (fd != -1) {
	res = read (fd, &memb_ring_id->seq, sizeof (uint64_t));
	close (fd);
	}
	/*
	* If file could not be opened or read, create a new ring id
	*/
	if ((fd == -1) \|\| (res != sizeof (uint64_t))) {
	memb_ring_id->seq = 0;
	umask(0);
	fd = open (filename, O_CREAT\|O_RDWR, 0700);
	if (fd != -1) {
	res = write (fd, &memb_ring_id->seq, sizeof (uint64_t));
	close (fd);
	if (res == -1) {
	LOGSYS_PERROR (errno, instance->totemsrp_log_level_warning,
	"Couldn't write ringid file '%s'", filename);
	}
	} else {
	LOGSYS_PERROR (errno, instance->totemsrp_log_level_warning,
	"Couldn't create ringid file '%s'", filename);
	}
	}

	totemip_copy(&memb_ring_id->rep, &instance->my_id.addr[0]);
	assert (!totemip_zero_check(&memb_ring_id->rep));
	instance->token_ring_id_seq = memb_ring_id->seq;
	}

	static void memb_ring_id_set_and_store (
	struct totemsrp_instance *instance,
	const struct memb_ring_id *ring_id)
	{
	char filename[256];
	int fd;
	int res;

	memcpy (&instance->my_ring_id, ring_id, sizeof (struct memb_ring_id));

	snprintf (filename, sizeof(filename), "%s/ringid_%s",
	rundir, totemip_print (&instance->my_id.addr[0]));

	fd = open (filename, O_WRONLY, 0777);
	if (fd == -1) {
	fd = open (filename, O_CREAT\|O_RDWR, 0777);
	}
	if (fd == -1) {
	LOGSYS_PERROR(errno, instance->totemsrp_log_level_warning,
	"Couldn't store new ring id %llx to stable storage",
	instance->my_ring_id.seq);
	assert (0);
	return;
	}
	log_printf (instance->totemsrp_log_level_debug,
	"Storing new sequence id for ring %llx\n", instance->my_ring_id.seq);
	//assert (fd > 0);
	res = write (fd, &instance->my_ring_id.seq, sizeof (unsigned long long));
	assert (res == sizeof (unsigned long long));
	close (fd);
	}

	int totemsrp_callback_token_create (
	void *srp_context,
	void **handle_out,
	enum totem_callback_token_type type,
	int delete,
	int (callback_fn) (enum totem_callback_token_type type, const void ),
	const void *data)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )srp_context;
	struct token_callback_instance *callback_handle;

	token_hold_cancel_send (instance);

	callback_handle = malloc (sizeof (struct token_callback_instance));
	if (callback_handle == 0) {
	return (-1);
	}
	handle_out = (void )callback_handle;
	list_init (&callback_handle->list);
	callback_handle->callback_fn = callback_fn;
	callback_handle->data = (void *) data;
	callback_handle->callback_type = type;
	callback_handle->delete = delete;
	switch (type) {
	case TOTEM_CALLBACK_TOKEN_RECEIVED:
	list_add (&callback_handle->list, &instance->token_callback_received_listhead);
	break;
	case TOTEM_CALLBACK_TOKEN_SENT:
	list_add (&callback_handle->list, &instance->token_callback_sent_listhead);
	break;
	}

	return (0);
	}

	void totemsrp_callback_token_destroy (void srp_context, void *handle_out)
	{
	struct token_callback_instance *h;

	if (*handle_out) {
	h = (struct token_callback_instance )handle_out;
	list_del (&h->list);
	free (h);
	h = NULL;
	*handle_out = 0;
	}
	}

	static void token_callbacks_execute (
	struct totemsrp_instance *instance,
	enum totem_callback_token_type type)
	{
	struct list_head *list;
	struct list_head *list_next;
	struct list_head *callback_listhead = 0;
	struct token_callback_instance *token_callback_instance;
	int res;
	int del;

	switch (type) {
	case TOTEM_CALLBACK_TOKEN_RECEIVED:
	callback_listhead = &instance->token_callback_received_listhead;
	break;
	case TOTEM_CALLBACK_TOKEN_SENT:
	callback_listhead = &instance->token_callback_sent_listhead;
	break;
	default:
	assert (0);
	}

	for (list = callback_listhead->next; list != callback_listhead;
	list = list_next) {

	token_callback_instance = list_entry (list, struct token_callback_instance, list);

	list_next = list->next;
	del = token_callback_instance->delete;
	if (del == 1) {
	list_del (list);
	}

	res = token_callback_instance->callback_fn (
	token_callback_instance->callback_type,
	token_callback_instance->data);
	/*
	* This callback failed to execute, try it again on the next token
	*/
	if (res == -1 && del == 1) {
	list_add (list, callback_listhead);
	} else if (del) {
	free (token_callback_instance);
	}
	}
	}

	/*
	* Flow control functions
	*/
	static unsigned int backlog_get (struct totemsrp_instance *instance)
	{
	unsigned int backlog = 0;

	if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
	backlog = cs_queue_used (&instance->new_message_queue);
	} else
	if (instance->memb_state == MEMB_STATE_RECOVERY) {
	backlog = cs_queue_used (&instance->retrans_message_queue);
	}
	instance->stats.token[instance->stats.latest_token].backlog_calc = backlog;
	return (backlog);
	}

	static int fcc_calculate (
	struct totemsrp_instance *instance,
	struct orf_token *token)
	{
	unsigned int transmits_allowed;
	unsigned int backlog_calc;

	transmits_allowed = instance->totem_config->max_messages;

	if (transmits_allowed > instance->totem_config->window_size - token->fcc) {
	transmits_allowed = instance->totem_config->window_size - token->fcc;
	}

	instance->my_cbl = backlog_get (instance);

	/*
	* Only do backlog calculation if there is a backlog otherwise
	* we would result in div by zero
	*/
	if (token->backlog + instance->my_cbl - instance->my_pbl) {
	backlog_calc = (instance->totem_config->window_size * instance->my_pbl) /
	(token->backlog + instance->my_cbl - instance->my_pbl);
	if (backlog_calc > 0 && transmits_allowed > backlog_calc) {
	transmits_allowed = backlog_calc;
	}
	}

	return (transmits_allowed);
	}

	/*
	* don't overflow the RTR sort queue
	*/
	static void fcc_rtr_limit (
	struct totemsrp_instance *instance,
	struct orf_token *token,
	unsigned int *transmits_allowed)
	{
	int check = QUEUE_RTR_ITEMS_SIZE_MAX;
	check -= (*transmits_allowed + instance->totem_config->window_size);
	assert (check >= 0);
	if (sq_lt_compare (instance->last_released +
	QUEUE_RTR_ITEMS_SIZE_MAX - *transmits_allowed -
	instance->totem_config->window_size,

	token->seq)) {

	*transmits_allowed = 0;
	}
	}

	static void fcc_token_update (
	struct totemsrp_instance *instance,
	struct orf_token *token,
	unsigned int msgs_transmitted)
	{
	token->fcc += msgs_transmitted - instance->my_trc;
	token->backlog += instance->my_cbl - instance->my_pbl;
	instance->my_trc = msgs_transmitted;
	instance->my_pbl = instance->my_cbl;
	}

	/*
	* Message Handlers
	*/

	unsigned long long int tv_old;
	/*
	* message handler called when TOKEN message type received
	*/
	static int message_handler_orf_token (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed)
	{
	char token_storage[1500];
	char token_convert[1500];
	struct orf_token *token = NULL;
	int forward_token;
	unsigned int transmits_allowed;
	unsigned int mcasted_retransmit;
	unsigned int mcasted_regular;
	unsigned int last_aru;

	#ifdef GIVEINFO
	unsigned long long tv_current;
	unsigned long long tv_diff;

	tv_current = qb_util_nano_current_get ();
	tv_diff = tv_current - tv_old;
	tv_old = tv_current;

	log_printf (instance->totemsrp_log_level_debug,
	"Time since last token %0.4f ms\n", ((float)tv_diff) / 1000000.0);
	#endif

	if (instance->orf_token_discard) {
	return (0);
	}
	#ifdef TEST_DROP_ORF_TOKEN_PERCENTAGE
	if (random()%100 < TEST_DROP_ORF_TOKEN_PERCENTAGE) {
	return (0);
	}
	#endif

	if (endian_conversion_needed) {
	orf_token_endian_convert ((struct orf_token *)msg,
	(struct orf_token *)token_convert);
	msg = (struct orf_token *)token_convert;
	}

	/*
	* Make copy of token and retransmit list in case we have
	* to flush incoming messages from the kernel queue
	*/
	token = (struct orf_token *)token_storage;
	memcpy (token, msg, sizeof (struct orf_token));
	memcpy (&token->rtr_list[0], (char *)msg + sizeof (struct orf_token),
	sizeof (struct rtr_item) * RETRANSMIT_ENTRIES_MAX);


	/*
	* Handle merge detection timeout
	*/
	if (token->seq == instance->my_last_seq) {
	start_merge_detect_timeout (instance);
	instance->my_seq_unchanged += 1;
	} else {
	cancel_merge_detect_timeout (instance);
	cancel_token_hold_retransmit_timeout (instance);
	instance->my_seq_unchanged = 0;
	}

	instance->my_last_seq = token->seq;

	#ifdef TEST_RECOVERY_MSG_COUNT
	if (instance->memb_state == MEMB_STATE_OPERATIONAL && token->seq > TEST_RECOVERY_MSG_COUNT) {
	return (0);
	}
	#endif

	totemrrp_recv_flush (instance->totemrrp_context);

	/*
	* Determine if we should hold (in reality drop) the token
	*/
	instance->my_token_held = 0;
	if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) &&
	instance->my_seq_unchanged > instance->totem_config->seqno_unchanged_const) {
	instance->my_token_held = 1;
	} else
	if (!totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) &&
	instance->my_seq_unchanged >= instance->totem_config->seqno_unchanged_const) {
	instance->my_token_held = 1;
	}

	/*
	* Hold onto token when there is no activity on ring and
	* this processor is the ring rep
	*/
	forward_token = 1;
	if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
	if (instance->my_token_held) {
	forward_token = 0;
	}
	}

	token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_RECEIVED);

	switch (instance->memb_state) {
	case MEMB_STATE_COMMIT:
	/* Discard token */
	break;

	case MEMB_STATE_OPERATIONAL:
	messages_free (instance, token->aru);
	/*
	* Do NOT add break, this case should also execute code in gather case.
	*/

	case MEMB_STATE_GATHER:
	/*
	* DO NOT add break, we use different free mechanism in recovery state
	*/

	case MEMB_STATE_RECOVERY:
	/*
	* Discard tokens from another configuration
	*/
	if (memcmp (&token->ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id)) != 0) {

	if ((forward_token)
	&& instance->use_heartbeat) {
	reset_heartbeat_timeout(instance);
	}
	else {
	cancel_heartbeat_timeout(instance);
	}

	return (0); /* discard token */
	}

	/*
	* Discard retransmitted tokens
	*/
	if (sq_lte_compare (token->token_seq, instance->my_token_seq)) {
	return (0); /* discard token */
	}
	last_aru = instance->my_last_aru;
	instance->my_last_aru = token->aru;

	transmits_allowed = fcc_calculate (instance, token);
	mcasted_retransmit = orf_token_rtr (instance, token, &transmits_allowed);

	fcc_rtr_limit (instance, token, &transmits_allowed);
	mcasted_regular = orf_token_mcast (instance, token, transmits_allowed);
	/*
	if (mcasted_regular) {
	printf ("mcasted regular %d\n", mcasted_regular);
	printf ("token seq %d\n", token->seq);
	}
	*/
	fcc_token_update (instance, token, mcasted_retransmit +
	mcasted_regular);

	if (sq_lt_compare (instance->my_aru, token->aru) \|\|
	instance->my_id.addr[0].nodeid == token->aru_addr \|\|
	token->aru_addr == 0) {

	token->aru = instance->my_aru;
	if (token->aru == token->seq) {
	token->aru_addr = 0;
	} else {
	token->aru_addr = instance->my_id.addr[0].nodeid;
	}
	}
	if (token->aru == last_aru && token->aru_addr != 0) {
	instance->my_aru_count += 1;
	} else {
	instance->my_aru_count = 0;
	}

	if (instance->my_aru_count > instance->totem_config->fail_to_recv_const &&
	token->aru_addr == instance->my_id.addr[0].nodeid) {

	log_printf (instance->totemsrp_log_level_error,
	"FAILED TO RECEIVE\n");

	instance->failed_to_recv = 1;

	memb_set_merge (&instance->my_id, 1,
	instance->my_failed_list,
	&instance->my_failed_list_entries);

	memb_state_gather_enter (instance, 6);
	} else {
	instance->my_token_seq = token->token_seq;
	token->token_seq += 1;

	if (instance->memb_state == MEMB_STATE_RECOVERY) {
	/*
	* instance->my_aru == instance->my_high_seq_received means this processor
	* has recovered all messages it can recover
	* (ie: its retrans queue is empty)
	*/
	if (cs_queue_is_empty (&instance->retrans_message_queue) == 0) {

	if (token->retrans_flg == 0) {
	token->retrans_flg = 1;
	instance->my_set_retrans_flg = 1;
	}
	} else
	if (token->retrans_flg == 1 && instance->my_set_retrans_flg) {
	token->retrans_flg = 0;
	instance->my_set_retrans_flg = 0;
	}
	log_printf (instance->totemsrp_log_level_debug,
	"token retrans flag is %d my set retrans flag%d retrans queue empty %d count %d, aru %x\n",
	token->retrans_flg, instance->my_set_retrans_flg,
	cs_queue_is_empty (&instance->retrans_message_queue),
	instance->my_retrans_flg_count, token->aru);
	if (token->retrans_flg == 0) {
	instance->my_retrans_flg_count += 1;
	} else {
	instance->my_retrans_flg_count = 0;
	}
	if (instance->my_retrans_flg_count == 2) {
	instance->my_install_seq = token->seq;
	}
	log_printf (instance->totemsrp_log_level_debug,
	"install seq %x aru %x high seq received %x\n",
	instance->my_install_seq, instance->my_aru, instance->my_high_seq_received);
	if (instance->my_retrans_flg_count >= 2 &&
	instance->my_received_flg == 0 &&
	sq_lte_compare (instance->my_install_seq, instance->my_aru)) {
	instance->my_received_flg = 1;
	instance->my_deliver_memb_entries = instance->my_trans_memb_entries;
	memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list,
	sizeof (struct totem_ip_address) * instance->my_trans_memb_entries);
	}
	if (instance->my_retrans_flg_count >= 3 &&
	sq_lte_compare (instance->my_install_seq, token->aru)) {
	instance->my_rotation_counter += 1;
	} else {
	instance->my_rotation_counter = 0;
	}
	if (instance->my_rotation_counter == 2) {
	log_printf (instance->totemsrp_log_level_debug,
	"retrans flag count %x token aru %x install seq %x aru %x %x\n",
	instance->my_retrans_flg_count, token->aru, instance->my_install_seq,
	instance->my_aru, token->seq);

	memb_state_operational_enter (instance);
	instance->my_rotation_counter = 0;
	instance->my_retrans_flg_count = 0;
	}
	}

	totemrrp_send_flush (instance->totemrrp_context);
	token_send (instance, token, forward_token);

	#ifdef GIVEINFO
	tv_current = qb_util_nano_current_get ();
	tv_diff = tv_current - tv_old;
	tv_old = tv_current;
	log_printf (instance->totemsrp_log_level_debug,
	"I held %0.4f ms\n",
	((float)tv_diff) / 1000000.0);
	#endif
	if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
	messages_deliver_to_app (instance, 0,
	instance->my_high_seq_received);
	}

	/*
	* Deliver messages after token has been transmitted
	* to improve performance
	*/
	reset_token_timeout (instance); // REVIEWED
	reset_token_retransmit_timeout (instance); // REVIEWED
	if (totemip_equal(&instance->my_id.addr[0], &instance->my_ring_id.rep) &&
	instance->my_token_held == 1) {

	start_token_hold_retransmit_timeout (instance);
	}

	token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_SENT);
	}
	break;
	}

	if ((forward_token)
	&& instance->use_heartbeat) {
	reset_heartbeat_timeout(instance);
	}
	else {
	cancel_heartbeat_timeout(instance);
	}

	return (0);
	}

	static void messages_deliver_to_app (
	struct totemsrp_instance *instance,
	int skip,
	unsigned int end_point)
	{
	struct sort_queue_item *sort_queue_item_p;
	unsigned int i;
	int res;
	struct mcast *mcast_in;
	struct mcast mcast_header;
	unsigned int range = 0;
	int endian_conversion_required;
	unsigned int my_high_delivered_stored = 0;


	range = end_point - instance->my_high_delivered;

	if (range) {
	log_printf (instance->totemsrp_log_level_debug,
	"Delivering %x to %x\n", instance->my_high_delivered,
	end_point);
	}
	assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
	my_high_delivered_stored = instance->my_high_delivered;

	/*
	* Deliver messages in order from rtr queue to pending delivery queue
	*/
	for (i = 1; i <= range; i++) {

	void *ptr = 0;

	/*
	* If out of range of sort queue, stop assembly
	*/
	res = sq_in_range (&instance->regular_sort_queue,
	my_high_delivered_stored + i);
	if (res == 0) {
	break;
	}

	res = sq_item_get (&instance->regular_sort_queue,
	my_high_delivered_stored + i, &ptr);
	/*
	* If hole, stop assembly
	*/
	if (res != 0 && skip == 0) {
	break;
	}

	instance->my_high_delivered = my_high_delivered_stored + i;

	if (res != 0) {
	continue;

	}

	sort_queue_item_p = ptr;

	mcast_in = sort_queue_item_p->mcast;
	assert (mcast_in != (struct mcast *)0xdeadbeef);

	endian_conversion_required = 0;
	if (mcast_in->header.endian_detector != ENDIAN_LOCAL) {
	endian_conversion_required = 1;
	mcast_endian_convert (mcast_in, &mcast_header);
	} else {
	memcpy (&mcast_header, mcast_in, sizeof (struct mcast));
	}

	/*
	* Skip messages not originated in instance->my_deliver_memb
	*/
	if (skip &&
	memb_set_subset (&mcast_header.system_from,
	1,
	instance->my_deliver_memb_list,
	instance->my_deliver_memb_entries) == 0) {

	instance->my_high_delivered = my_high_delivered_stored + i;

	continue;
	}

	/*
	* Message found
	*/
	log_printf (instance->totemsrp_log_level_debug,
	"Delivering MCAST message with seq %x to pending delivery queue\n",
	mcast_header.seq);

	/*
	* Message is locally originated multicast
	*/
	instance->totemsrp_deliver_fn (
	mcast_header.header.nodeid,
	((char *)sort_queue_item_p->mcast) + sizeof (struct mcast),
	sort_queue_item_p->msg_len - sizeof (struct mcast),
	endian_conversion_required);
	}
	}

	/*
	* recv message handler called when MCAST message type received
	*/
	static int message_handler_mcast (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed)
	{
	struct sort_queue_item sort_queue_item;
	struct sq *sort_queue;
	struct mcast mcast_header;


	if (endian_conversion_needed) {
	mcast_endian_convert (msg, &mcast_header);
	} else {
	memcpy (&mcast_header, msg, sizeof (struct mcast));
	}

	if (mcast_header.header.encapsulated == MESSAGE_ENCAPSULATED) {
	sort_queue = &instance->recovery_sort_queue;
	} else {
	sort_queue = &instance->regular_sort_queue;
	}

	assert (msg_len <= FRAME_SIZE_MAX);

	#ifdef TEST_DROP_MCAST_PERCENTAGE
	if (random()%100 < TEST_DROP_MCAST_PERCENTAGE) {
	return (0);
	}
	#endif

	/*
	* If the message is foreign execute the switch below
	*/
	if (memcmp (&instance->my_ring_id, &mcast_header.ring_id,
	sizeof (struct memb_ring_id)) != 0) {

	switch (instance->memb_state) {
	case MEMB_STATE_OPERATIONAL:
	memb_set_merge (
	&mcast_header.system_from, 1,
	instance->my_proc_list, &instance->my_proc_list_entries);
	memb_state_gather_enter (instance, 7);
	break;

	case MEMB_STATE_GATHER:
	if (!memb_set_subset (
	&mcast_header.system_from,
	1,
	instance->my_proc_list,
	instance->my_proc_list_entries)) {

	memb_set_merge (&mcast_header.system_from, 1,
	instance->my_proc_list, &instance->my_proc_list_entries);
	memb_state_gather_enter (instance, 8);
	return (0);
	}
	break;

	case MEMB_STATE_COMMIT:
	/* discard message */
	instance->stats.rx_msg_dropped++;
	break;

	case MEMB_STATE_RECOVERY:
	/* discard message */
	instance->stats.rx_msg_dropped++;
	break;
	}
	return (0);
	}

	log_printf (instance->totemsrp_log_level_debug,
	"Received ringid(%s:%lld) seq %x\n",
	totemip_print (&mcast_header.ring_id.rep),
	mcast_header.ring_id.seq,
	mcast_header.seq);

	/*
	* Add mcast message to rtr queue if not already in rtr queue
	* otherwise free io vectors
	*/
	if (msg_len > 0 && msg_len <= FRAME_SIZE_MAX &&
	sq_in_range (sort_queue, mcast_header.seq) &&
	sq_item_inuse (sort_queue, mcast_header.seq) == 0) {

	/*
	* Allocate new multicast memory block
	*/
	// TODO LEAK
	sort_queue_item.mcast = totemsrp_buffer_alloc (instance);
	if (sort_queue_item.mcast == NULL) {
	return (-1); /* error here is corrected by the algorithm */
	}
	memcpy (sort_queue_item.mcast, msg, msg_len);
	sort_queue_item.msg_len = msg_len;

	if (sq_lt_compare (instance->my_high_seq_received,
	mcast_header.seq)) {
	instance->my_high_seq_received = mcast_header.seq;
	}

	sq_item_add (sort_queue, &sort_queue_item, mcast_header.seq);
	}

	update_aru (instance);
	if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
	messages_deliver_to_app (instance, 0, instance->my_high_seq_received);
	}

	/* TODO remove from retrans message queue for old ring in recovery state */
	return (0);
	}

	static int message_handler_memb_merge_detect (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed)
	{
	struct memb_merge_detect memb_merge_detect;


	if (endian_conversion_needed) {
	memb_merge_detect_endian_convert (msg, &memb_merge_detect);
	} else {
	memcpy (&memb_merge_detect, msg,
	sizeof (struct memb_merge_detect));
	}

	/*
	* do nothing if this is a merge detect from this configuration
	*/
	if (memcmp (&instance->my_ring_id, &memb_merge_detect.ring_id,
	sizeof (struct memb_ring_id)) == 0) {

	return (0);
	}

	/*
	* Execute merge operation
	*/
	switch (instance->memb_state) {
	case MEMB_STATE_OPERATIONAL:
	memb_set_merge (&memb_merge_detect.system_from, 1,
	instance->my_proc_list, &instance->my_proc_list_entries);
	memb_state_gather_enter (instance, 9);
	break;

	case MEMB_STATE_GATHER:
	if (!memb_set_subset (
	&memb_merge_detect.system_from,
	1,
	instance->my_proc_list,
	instance->my_proc_list_entries)) {

	memb_set_merge (&memb_merge_detect.system_from, 1,
	instance->my_proc_list, &instance->my_proc_list_entries);
	memb_state_gather_enter (instance, 10);
	return (0);
	}
	break;

	case MEMB_STATE_COMMIT:
	/* do nothing in commit */
	break;

	case MEMB_STATE_RECOVERY:
	/* do nothing in recovery */
	break;
	}
	return (0);
	}

	static void memb_join_process (
	struct totemsrp_instance *instance,
	const struct memb_join *memb_join)
	{
	struct srp_addr *proc_list;
	struct srp_addr *failed_list;
	int gather_entered = 0;
	int fail_minus_memb_entries = 0;
	struct srp_addr fail_minus_memb[PROCESSOR_COUNT_MAX];

	proc_list = (struct srp_addr *)memb_join->end_of_memb_join;
	failed_list = proc_list + memb_join->proc_list_entries;

	/*
	memb_set_print ("proclist", proc_list, memb_join->proc_list_entries);
	memb_set_print ("faillist", failed_list, memb_join->failed_list_entries);
	memb_set_print ("my_proclist", instance->my_proc_list, instance->my_proc_list_entries);
	memb_set_print ("my_faillist", instance->my_failed_list, instance->my_failed_list_entries);
	-*/

	if (memb_set_equal (proc_list,
	memb_join->proc_list_entries,
	instance->my_proc_list,
	instance->my_proc_list_entries) &&

	memb_set_equal (failed_list,
	memb_join->failed_list_entries,
	instance->my_failed_list,
	instance->my_failed_list_entries)) {

	memb_consensus_set (instance, &memb_join->system_from);

	if (memb_consensus_agreed (instance) && instance->failed_to_recv == 1) {
	instance->failed_to_recv = 0;
	srp_addr_copy (&instance->my_proc_list[0],
	&instance->my_id);
	instance->my_proc_list_entries = 1;
	instance->my_failed_list_entries = 0;

	memb_state_commit_token_create (instance);

	memb_state_commit_enter (instance);
	return;
	}
	if (memb_consensus_agreed (instance) &&
	memb_lowest_in_config (instance)) {

	memb_state_commit_token_create (instance);

	memb_state_commit_enter (instance);
	} else {
	return;
	}
	} else
	if (memb_set_subset (proc_list,
	memb_join->proc_list_entries,
	instance->my_proc_list,
	instance->my_proc_list_entries) &&

	memb_set_subset (failed_list,
	memb_join->failed_list_entries,
	instance->my_failed_list,
	instance->my_failed_list_entries)) {

	return;
	} else
	if (memb_set_subset (&memb_join->system_from, 1,
	instance->my_failed_list, instance->my_failed_list_entries)) {

	return;
	} else {
	memb_set_merge (proc_list,
	memb_join->proc_list_entries,
	instance->my_proc_list, &instance->my_proc_list_entries);

	if (memb_set_subset (
	&instance->my_id, 1,
	failed_list, memb_join->failed_list_entries)) {

	memb_set_merge (
	&memb_join->system_from, 1,
	instance->my_failed_list, &instance->my_failed_list_entries);
	} else {
	if (memb_set_subset (
	&memb_join->system_from, 1,
	instance->my_memb_list,
	instance->my_memb_entries)) {

	if (memb_set_subset (
	&memb_join->system_from, 1,
	instance->my_failed_list,
	instance->my_failed_list_entries) == 0) {

	memb_set_merge (failed_list,
	memb_join->failed_list_entries,
	instance->my_failed_list, &instance->my_failed_list_entries);
	} else {
	memb_set_subtract (fail_minus_memb,
	&fail_minus_memb_entries,
	failed_list,
	memb_join->failed_list_entries,
	instance->my_memb_list,
	instance->my_memb_entries);

	memb_set_merge (fail_minus_memb,
	fail_minus_memb_entries,
	instance->my_failed_list,
	&instance->my_failed_list_entries);
	}
	}
	}
	memb_state_gather_enter (instance, 11);
	gather_entered = 1;
	}
	if (gather_entered == 0 &&
	instance->memb_state == MEMB_STATE_OPERATIONAL) {

	memb_state_gather_enter (instance, 12);
	}
	}

	static void memb_join_endian_convert (const struct memb_join in, struct memb_join out)
	{
	int i;
	struct srp_addr *in_proc_list;
	struct srp_addr *in_failed_list;
	struct srp_addr *out_proc_list;
	struct srp_addr *out_failed_list;

	out->header.type = in->header.type;
	out->header.endian_detector = ENDIAN_LOCAL;
	out->header.nodeid = swab32 (in->header.nodeid);
	srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
	out->proc_list_entries = swab32 (in->proc_list_entries);
	out->failed_list_entries = swab32 (in->failed_list_entries);
	out->ring_seq = swab64 (in->ring_seq);

	in_proc_list = (struct srp_addr *)in->end_of_memb_join;
	in_failed_list = in_proc_list + out->proc_list_entries;
	out_proc_list = (struct srp_addr *)out->end_of_memb_join;
	out_failed_list = out_proc_list + out->proc_list_entries;

	for (i = 0; i < out->proc_list_entries; i++) {
	srp_addr_copy_endian_convert (&out_proc_list[i], &in_proc_list[i]);
	}
	for (i = 0; i < out->failed_list_entries; i++) {
	srp_addr_copy_endian_convert (&out_failed_list[i], &in_failed_list[i]);
	}
	}

	static void memb_commit_token_endian_convert (const struct memb_commit_token in, struct memb_commit_token out)
	{
	int i;
	struct srp_addr in_addr = (struct srp_addr )in->end_of_commit_token;
	struct srp_addr out_addr = (struct srp_addr )out->end_of_commit_token;
	struct memb_commit_token_memb_entry *in_memb_list;
	struct memb_commit_token_memb_entry *out_memb_list;

	out->header.type = in->header.type;
	out->header.endian_detector = ENDIAN_LOCAL;
	out->header.nodeid = swab32 (in->header.nodeid);
	out->token_seq = swab32 (in->token_seq);
	totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep);
	out->ring_id.seq = swab64 (in->ring_id.seq);
	out->retrans_flg = swab32 (in->retrans_flg);
	out->memb_index = swab32 (in->memb_index);
	out->addr_entries = swab32 (in->addr_entries);

	in_memb_list = (struct memb_commit_token_memb_entry *)(in_addr + out->addr_entries);
	out_memb_list = (struct memb_commit_token_memb_entry *)(out_addr + out->addr_entries);
	for (i = 0; i < out->addr_entries; i++) {
	srp_addr_copy_endian_convert (&out_addr[i], &in_addr[i]);

	/*
	* Only convert the memb entry if it has been set
	*/
	if (in_memb_list[i].ring_id.rep.family != 0) {
	totemip_copy_endian_convert (&out_memb_list[i].ring_id.rep,
	&in_memb_list[i].ring_id.rep);

	out_memb_list[i].ring_id.seq =
	swab64 (in_memb_list[i].ring_id.seq);
	out_memb_list[i].aru = swab32 (in_memb_list[i].aru);
	out_memb_list[i].high_delivered = swab32 (in_memb_list[i].high_delivered);
	out_memb_list[i].received_flg = swab32 (in_memb_list[i].received_flg);
	}
	}
	}

	static void orf_token_endian_convert (const struct orf_token in, struct orf_token out)
	{
	int i;

	out->header.type = in->header.type;
	out->header.endian_detector = ENDIAN_LOCAL;
	out->header.nodeid = swab32 (in->header.nodeid);
	out->seq = swab32 (in->seq);
	out->token_seq = swab32 (in->token_seq);
	out->aru = swab32 (in->aru);
	totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep);
	out->aru_addr = swab32(in->aru_addr);
	out->ring_id.seq = swab64 (in->ring_id.seq);
	out->fcc = swab32 (in->fcc);
	out->backlog = swab32 (in->backlog);
	out->retrans_flg = swab32 (in->retrans_flg);
	out->rtr_list_entries = swab32 (in->rtr_list_entries);
	for (i = 0; i < out->rtr_list_entries; i++) {
	totemip_copy_endian_convert(&out->rtr_list[i].ring_id.rep, &in->rtr_list[i].ring_id.rep);
	out->rtr_list[i].ring_id.seq = swab64 (in->rtr_list[i].ring_id.seq);
	out->rtr_list[i].seq = swab32 (in->rtr_list[i].seq);
	}
	}

	static void mcast_endian_convert (const struct mcast in, struct mcast out)
	{
	out->header.type = in->header.type;
	out->header.endian_detector = ENDIAN_LOCAL;
	out->header.nodeid = swab32 (in->header.nodeid);
	out->header.encapsulated = in->header.encapsulated;

	out->seq = swab32 (in->seq);
	out->this_seqno = swab32 (in->this_seqno);
	totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep);
	out->ring_id.seq = swab64 (in->ring_id.seq);
	out->node_id = swab32 (in->node_id);
	out->guarantee = swab32 (in->guarantee);
	srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
	}

	static void memb_merge_detect_endian_convert (
	const struct memb_merge_detect *in,
	struct memb_merge_detect *out)
	{
	out->header.type = in->header.type;
	out->header.endian_detector = ENDIAN_LOCAL;
	out->header.nodeid = swab32 (in->header.nodeid);
	totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep);
	out->ring_id.seq = swab64 (in->ring_id.seq);
	srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
	}

	static int message_handler_memb_join (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed)
	{
	const struct memb_join *memb_join;
	struct memb_join *memb_join_convert = alloca (msg_len);

	if (endian_conversion_needed) {
	memb_join = memb_join_convert;
	memb_join_endian_convert (msg, memb_join_convert);

	} else {
	memb_join = msg;
	}
	/*
	* If the process paused because it wasn't scheduled in a timely
	* fashion, flush the join messages because they may be queued
	* entries
	*/
	if (pause_flush (instance)) {
	return (0);
	}

	if (instance->token_ring_id_seq < memb_join->ring_seq) {
	instance->token_ring_id_seq = memb_join->ring_seq;
	}
	switch (instance->memb_state) {
	case MEMB_STATE_OPERATIONAL:
	memb_join_process (instance, memb_join);
	break;

	case MEMB_STATE_GATHER:
	memb_join_process (instance, memb_join);
	break;

	case MEMB_STATE_COMMIT:
	if (memb_set_subset (&memb_join->system_from,
	1,
	instance->my_new_memb_list,
	instance->my_new_memb_entries) &&

	memb_join->ring_seq >= instance->my_ring_id.seq) {

	memb_join_process (instance, memb_join);
	memb_state_gather_enter (instance, 13);
	}
	break;

	case MEMB_STATE_RECOVERY:
	if (memb_set_subset (&memb_join->system_from,
	1,
	instance->my_new_memb_list,
	instance->my_new_memb_entries) &&

	memb_join->ring_seq >= instance->my_ring_id.seq) {

	memb_join_process (instance, memb_join);
	memb_recovery_state_token_loss (instance);
	memb_state_gather_enter (instance, 14);
	}
	break;
	}
	return (0);
	}

	static int message_handler_memb_commit_token (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed)
	{
	struct memb_commit_token *memb_commit_token_convert = alloca (msg_len);
	struct memb_commit_token *memb_commit_token;
	struct srp_addr sub[PROCESSOR_COUNT_MAX];
	int sub_entries;

	struct srp_addr *addr;

	log_printf (instance->totemsrp_log_level_debug,
	"got commit token\n");

	if (endian_conversion_needed) {
	memb_commit_token_endian_convert (msg, memb_commit_token_convert);
	} else {
	memcpy (memb_commit_token_convert, msg, msg_len);
	}
	memb_commit_token = memb_commit_token_convert;
	addr = (struct srp_addr *)memb_commit_token->end_of_commit_token;

	#ifdef TEST_DROP_COMMIT_TOKEN_PERCENTAGE
	if (random()%100 < TEST_DROP_COMMIT_TOKEN_PERCENTAGE) {
	return (0);
	}
	#endif
	switch (instance->memb_state) {
	case MEMB_STATE_OPERATIONAL:
	/* discard token */
	break;

	case MEMB_STATE_GATHER:
	memb_set_subtract (sub, &sub_entries,
	instance->my_proc_list, instance->my_proc_list_entries,
	instance->my_failed_list, instance->my_failed_list_entries);

	if (memb_set_equal (addr,
	memb_commit_token->addr_entries,
	sub,
	sub_entries) &&

	memb_commit_token->ring_id.seq > instance->my_ring_id.seq) {
	memcpy (instance->commit_token, memb_commit_token, msg_len);
	memb_state_commit_enter (instance);
	}
	break;

	case MEMB_STATE_COMMIT:
	/*
	* If retransmitted commit tokens are sent on this ring
	* filter them out and only enter recovery once the
	* commit token has traversed the array. This is
	* determined by :
	* memb_commit_token->memb_index == memb_commit_token->addr_entries) {
	*/
	if (memb_commit_token->ring_id.seq == instance->my_ring_id.seq &&
	memb_commit_token->memb_index == memb_commit_token->addr_entries) {
	memb_state_recovery_enter (instance, memb_commit_token);
	}
	break;

	case MEMB_STATE_RECOVERY:
	if (totemip_equal (&instance->my_id.addr[0], &instance->my_ring_id.rep)) {
	log_printf (instance->totemsrp_log_level_debug,
	"Sending initial ORF token\n");

	// TODO convert instead of initiate
	orf_token_send_initial (instance);
	reset_token_timeout (instance); // REVIEWED
	reset_token_retransmit_timeout (instance); // REVIEWED
	}
	break;
	}
	return (0);
	}

	static int message_handler_token_hold_cancel (
	struct totemsrp_instance *instance,
	const void *msg,
	size_t msg_len,
	int endian_conversion_needed)
	{
	const struct token_hold_cancel *token_hold_cancel = msg;

	if (memcmp (&token_hold_cancel->ring_id, &instance->my_ring_id,
	sizeof (struct memb_ring_id)) == 0) {

	instance->my_seq_unchanged = 0;
	if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
	timer_function_token_retransmit_timeout (instance);
	}
	}
	return (0);
	}

	void main_deliver_fn (
	void *context,
	const void *msg,
	unsigned int msg_len)
	{
	struct totemsrp_instance *instance = context;
	const struct message_header *message_header = msg;

	if (msg_len < sizeof (struct message_header)) {
	log_printf (instance->totemsrp_log_level_security,
	"Received message is too short... ignoring %u.\n",
	(unsigned int)msg_len);
	return;
	}

	switch (message_header->type) {
	case MESSAGE_TYPE_ORF_TOKEN:
	instance->stats.orf_token_rx++;
	break;
	case MESSAGE_TYPE_MCAST:
	instance->stats.mcast_rx++;
	break;
	case MESSAGE_TYPE_MEMB_MERGE_DETECT:
	instance->stats.memb_merge_detect_rx++;
	break;
	case MESSAGE_TYPE_MEMB_JOIN:
	instance->stats.memb_join_rx++;
	break;
	case MESSAGE_TYPE_MEMB_COMMIT_TOKEN:
	instance->stats.memb_commit_token_rx++;
	break;
	case MESSAGE_TYPE_TOKEN_HOLD_CANCEL:
	instance->stats.token_hold_cancel_rx++;
	break;
	default:
	log_printf (instance->totemsrp_log_level_security, "Type of received message is wrong... ignoring %d.\n", (int)message_header->type);
	printf ("wrong message type\n");
	instance->stats.rx_msg_dropped++;
	return;
	}
	/*
	* Handle incoming message
	*/
	totemsrp_message_handlers.handler_functions[(int)message_header->type] (
	instance,
	msg,
	msg_len,
	message_header->endian_detector != ENDIAN_LOCAL);
	}

	void main_iface_change_fn (
	void *context,
	const struct totem_ip_address *iface_addr,
	unsigned int iface_no)
	{
	struct totemsrp_instance *instance = context;
	int i;

	totemip_copy (&instance->my_id.addr[iface_no], iface_addr);
	assert (instance->my_id.addr[iface_no].nodeid);

	totemip_copy (&instance->my_memb_list[0].addr[iface_no], iface_addr);

	if (instance->iface_changes++ == 0) {
	memb_ring_id_create_or_load (instance, &instance->my_ring_id);
	log_printf (
	instance->totemsrp_log_level_debug,
	"Created or loaded sequence id %lld.%s for this ring.\n",
	instance->my_ring_id.seq,
	totemip_print (&instance->my_ring_id.rep));

	if (instance->totemsrp_service_ready_fn) {
	instance->totemsrp_service_ready_fn ();
	}

	}

	for (i = 0; i < instance->totem_config->interfaces[iface_no].member_count; i++) {
	totemsrp_member_add (instance,
	&instance->totem_config->interfaces[iface_no].member_list[i],
	iface_no);
	}

	if (instance->iface_changes >= instance->totem_config->interface_count) {
	memb_state_gather_enter (instance, 15);
	}
	}

	void totemsrp_net_mtu_adjust (struct totem_config *totem_config) {
	totem_config->net_mtu -= sizeof (struct mcast);
	}

	void totemsrp_service_ready_register (
	void *context,
	void (*totem_service_ready) (void))
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )context;

	instance->totemsrp_service_ready_fn = totem_service_ready;
	}

	int totemsrp_member_add (
	void *context,
	const struct totem_ip_address *member,
	int ring_no)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )context;
	int res;

	res = totemrrp_member_add (instance->totemrrp_context, member, ring_no);

	return (res);
	}

	int totemsrp_member_remove (
	void *context,
	const struct totem_ip_address *member,
	int ring_no)
	{
	struct totemsrp_instance instance = (struct totemsrp_instance )context;
	int res;

	res = totemrrp_member_remove (instance->totemrrp_context, member, ring_no);

	return (res);
	}
	diff --git a/services/cpg.c b/services/cpg.c
	index 1b8d5b45..e6001d1a 100644
	--- a/services/cpg.c
	+++ b/services/cpg.c
	@@ -1,2052 +1,2052 @@
	/*
	* Copyright (c) 2006-2009 Red Hat, Inc.
	*
	* All rights reserved.
	*
	* Author: Christine Caulfield (ccaulfie@redhat.com)
	* Author: Jan Friesse (jfriesse@redhat.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 CONTIBUTORS "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.
	*/

	#include <config.h>

	#ifdef HAVE_ALLOCA_H
	#include <alloca.h>
	#endif
	#include <sys/types.h>
	#include <sys/socket.h>
	#include <sys/un.h>
	#include <sys/ioctl.h>
	#include <netinet/in.h>
	#include <sys/uio.h>
	#include <unistd.h>
	#include <fcntl.h>
	#include <stdlib.h>
	#include <stdio.h>
	#include <errno.h>
	#include <time.h>
	#include <assert.h>
	#include <unistd.h>
	#include <netinet/in.h>
	#include <arpa/inet.h>
	#include <sys/mman.h>

	#include <corosync/corotypes.h>
	#include <qb/qbipc_common.h>
	#include <corosync/corodefs.h>
	#include <corosync/list.h>
	#include <corosync/jhash.h>
	#include <corosync/lcr/lcr_comp.h>
	#include <corosync/engine/logsys.h>
	#include <corosync/engine/coroapi.h>

	#include <corosync/cpg.h>
	#include <corosync/ipc_cpg.h>

	LOGSYS_DECLARE_SUBSYS ("CPG");

	#define GROUP_HASH_SIZE 32

	enum cpg_message_req_types {
	MESSAGE_REQ_EXEC_CPG_PROCJOIN = 0,
	MESSAGE_REQ_EXEC_CPG_PROCLEAVE = 1,
	MESSAGE_REQ_EXEC_CPG_JOINLIST = 2,
	MESSAGE_REQ_EXEC_CPG_MCAST = 3,
	MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD = 4,
	MESSAGE_REQ_EXEC_CPG_DOWNLIST = 5
	};

	struct zcb_mapped {
	struct list_head list;
	void *addr;
	size_t size;
	};
	/*
	* state` exec deliver
	* match group name, pid -> if matched deliver for YES:
	* XXX indicates impossible state
	*
	* join leave mcast
	* UNJOINED XXX XXX NO
	* LEAVE_STARTED XXX YES(unjoined_enter) YES
	* JOIN_STARTED YES(join_started_enter) XXX NO
	* JOIN_COMPLETED XXX NO YES
	*
	* join_started_enter
	* set JOIN_COMPLETED
	* add entry to process_info list
	* unjoined_enter
	* set UNJOINED
	* delete entry from process_info list
	*
	*
	* library accept join error codes
	* UNJOINED YES(CS_OK) set JOIN_STARTED
	* LEAVE_STARTED NO(CS_ERR_BUSY)
	* JOIN_STARTED NO(CS_ERR_EXIST)
	* JOIN_COMPlETED NO(CS_ERR_EXIST)
	*
	* library accept leave error codes
	* UNJOINED NO(CS_ERR_NOT_EXIST)
	* LEAVE_STARTED NO(CS_ERR_NOT_EXIST)
	* JOIN_STARTED NO(CS_ERR_BUSY)
	* JOIN_COMPLETED YES(CS_OK) set LEAVE_STARTED
	*
	* library accept mcast
	* UNJOINED NO(CS_ERR_NOT_EXIST)
	* LEAVE_STARTED NO(CS_ERR_NOT_EXIST)
	* JOIN_STARTED YES(CS_OK)
	* JOIN_COMPLETED YES(CS_OK)
	*/
	enum cpd_state {
	CPD_STATE_UNJOINED,
	CPD_STATE_LEAVE_STARTED,
	CPD_STATE_JOIN_STARTED,
	CPD_STATE_JOIN_COMPLETED
	};

	enum cpg_sync_state {
	CPGSYNC_DOWNLIST,
	CPGSYNC_JOINLIST
	};

	enum cpg_downlist_state_e {
	CPG_DOWNLIST_NONE,
	CPG_DOWNLIST_WAITING_FOR_MESSAGES,
	CPG_DOWNLIST_APPLYING,
	};
	static enum cpg_downlist_state_e downlist_state;
	static struct list_head downlist_messages_head;

	struct cpg_pd {
	void *conn;
	mar_cpg_name_t group_name;
	uint32_t pid;
	enum cpd_state cpd_state;
	unsigned int flags;
	int initial_totem_conf_sent;
	struct list_head list;
	struct list_head iteration_instance_list_head;
	struct list_head zcb_mapped_list_head;
	};

	struct cpg_iteration_instance {
	hdb_handle_t handle;
	struct list_head list;
	struct list_head items_list_head; /* List of process_info */
	struct list_head *current_pointer;
	};

	DECLARE_HDB_DATABASE(cpg_iteration_handle_t_db,NULL);

	DECLARE_LIST_INIT(cpg_pd_list_head);

	static unsigned int my_member_list[PROCESSOR_COUNT_MAX];

	static unsigned int my_member_list_entries;

	static unsigned int my_old_member_list[PROCESSOR_COUNT_MAX];

	static unsigned int my_old_member_list_entries = 0;

	static struct corosync_api_v1 *api = NULL;

	static enum cpg_sync_state my_sync_state = CPGSYNC_DOWNLIST;

	static mar_cpg_ring_id_t last_sync_ring_id;

	struct process_info {
	unsigned int nodeid;
	uint32_t pid;
	mar_cpg_name_t group;
	struct list_head list; /* on the group_info members list */
	};
	DECLARE_LIST_INIT(process_info_list_head);

	struct join_list_entry {
	uint32_t pid;
	mar_cpg_name_t group_name;
	};

	/*
	* Service Interfaces required by service_message_handler struct
	*/
	static int cpg_exec_init_fn (struct corosync_api_v1 *);

	static int cpg_lib_init_fn (void *conn);

	static int cpg_lib_exit_fn (void *conn);

	static void message_handler_req_exec_cpg_procjoin (
	const void *message,
	unsigned int nodeid);

	static void message_handler_req_exec_cpg_procleave (
	const void *message,
	unsigned int nodeid);

	static void message_handler_req_exec_cpg_joinlist (
	const void *message,
	unsigned int nodeid);

	static void message_handler_req_exec_cpg_mcast (
	const void *message,
	unsigned int nodeid);

	static void message_handler_req_exec_cpg_downlist_old (
	const void *message,
	unsigned int nodeid);

	static void message_handler_req_exec_cpg_downlist (
	const void *message,
	unsigned int nodeid);

	static void exec_cpg_procjoin_endian_convert (void *msg);

	static void exec_cpg_joinlist_endian_convert (void *msg);

	static void exec_cpg_mcast_endian_convert (void *msg);

	static void exec_cpg_downlist_endian_convert_old (void *msg);

	static void exec_cpg_downlist_endian_convert (void *msg);

	static void message_handler_req_lib_cpg_join (void conn, const void message);

	static void message_handler_req_lib_cpg_leave (void conn, const void message);

	static void message_handler_req_lib_cpg_finalize (void conn, const void message);

	static void message_handler_req_lib_cpg_mcast (void conn, const void message);

	static void message_handler_req_lib_cpg_membership (void *conn,
	const void *message);

	static void message_handler_req_lib_cpg_local_get (void *conn,
	const void *message);

	static void message_handler_req_lib_cpg_iteration_initialize (
	void *conn,
	const void *message);

	static void message_handler_req_lib_cpg_iteration_next (
	void *conn,
	const void *message);

	static void message_handler_req_lib_cpg_iteration_finalize (
	void *conn,
	const void *message);

	static void message_handler_req_lib_cpg_zc_alloc (
	void *conn,
	const void *message);

	static void message_handler_req_lib_cpg_zc_free (
	void *conn,
	const void *message);

	static void message_handler_req_lib_cpg_zc_execute (
	void *conn,
	const void *message);

	static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason);

	static int cpg_exec_send_downlist(void);

	static int cpg_exec_send_joinlist(void);

	static void downlist_messages_delete (void);

	static void downlist_master_choose_and_send (void);

	static void cpg_sync_init_v2 (
	const unsigned int *trans_list,
	size_t trans_list_entries,
	const unsigned int *member_list,
	size_t member_list_entries,
	const struct memb_ring_id *ring_id);

	static int cpg_sync_process (void);

	static void cpg_sync_activate (void);

	static void cpg_sync_abort (void);

	static int notify_lib_totem_membership (
	void *conn,
	int member_list_entries,
	const unsigned int *member_list);

	static inline int zcb_all_free (
	struct cpg_pd *cpd);

	/*
	* Library Handler Definition
	*/
	static struct corosync_lib_handler cpg_lib_engine[] =
	{
	- { /* 0 */
	+ { /* 0 - MESSAGE_REQ_CPG_JOIN */
	.lib_handler_fn = message_handler_req_lib_cpg_join,
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
	},
	- { /* 1 */
	+ { /* 1 - MESSAGE_REQ_CPG_LEAVE */
	.lib_handler_fn = message_handler_req_lib_cpg_leave,
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
	},
	- { /* 2 */
	+ { /* 2 - MESSAGE_REQ_CPG_MCAST */
	.lib_handler_fn = message_handler_req_lib_cpg_mcast,
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
	},
	- { /* 3 */
	+ { /* 3 - MESSAGE_REQ_CPG_MEMBERSHIP */
	.lib_handler_fn = message_handler_req_lib_cpg_membership,
	.flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
	},
	- { /* 4 */
	+ { /* 4 - MESSAGE_REQ_CPG_LOCAL_GET */
	.lib_handler_fn = message_handler_req_lib_cpg_local_get,
	.flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
	},
	- { /* 5 */
	+ { /* 5 - MESSAGE_REQ_CPG_ITERATIONINITIALIZE */
	.lib_handler_fn = message_handler_req_lib_cpg_iteration_initialize,
	.flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
	},
	- { /* 6 */
	+ { /* 6 - MESSAGE_REQ_CPG_ITERATIONNEXT */
	.lib_handler_fn = message_handler_req_lib_cpg_iteration_next,
	.flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
	},
	- { /* 7 */
	+ { /* 7 - MESSAGE_REQ_CPG_ITERATIONFINALIZE */
	.lib_handler_fn = message_handler_req_lib_cpg_iteration_finalize,
	.flow_control = CS_LIB_FLOW_CONTROL_NOT_REQUIRED
	},
	- { /* 8 */
	+ { /* 8 - MESSAGE_REQ_CPG_FINALIZE */
	.lib_handler_fn = message_handler_req_lib_cpg_finalize,
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
	},
	{ /* 9 */
	.lib_handler_fn = message_handler_req_lib_cpg_zc_alloc,
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
	},
	{ /* 10 */
	.lib_handler_fn = message_handler_req_lib_cpg_zc_free,
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
	},
	{ /* 11 */
	.lib_handler_fn = message_handler_req_lib_cpg_zc_execute,
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED
	},


	};

	static struct corosync_exec_handler cpg_exec_engine[] =
	{
	- { /* 0 */
	+ { /* 0 - MESSAGE_REQ_EXEC_CPG_PROCJOIN */
	.exec_handler_fn = message_handler_req_exec_cpg_procjoin,
	.exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
	},
	- { /* 1 */
	+ { /* 1 - MESSAGE_REQ_EXEC_CPG_PROCLEAVE */
	.exec_handler_fn = message_handler_req_exec_cpg_procleave,
	.exec_endian_convert_fn = exec_cpg_procjoin_endian_convert
	},
	- { /* 2 */
	+ { /* 2 - MESSAGE_REQ_EXEC_CPG_JOINLIST */
	.exec_handler_fn = message_handler_req_exec_cpg_joinlist,
	.exec_endian_convert_fn = exec_cpg_joinlist_endian_convert
	},
	- { /* 3 */
	+ { /* 3 - MESSAGE_REQ_EXEC_CPG_MCAST */
	.exec_handler_fn = message_handler_req_exec_cpg_mcast,
	.exec_endian_convert_fn = exec_cpg_mcast_endian_convert
	},
	- { /* 4 */
	+ { /* 4 - MESSAGE_REQ_EXEC_CPG_DOWNLIST_OLD */
	.exec_handler_fn = message_handler_req_exec_cpg_downlist_old,
	.exec_endian_convert_fn = exec_cpg_downlist_endian_convert_old
	},
	- { /* 5 */
	+ { /* 5 - MESSAGE_REQ_EXEC_CPG_DOWNLIST */
	.exec_handler_fn = message_handler_req_exec_cpg_downlist,
	.exec_endian_convert_fn = exec_cpg_downlist_endian_convert
	},
	};

	struct corosync_service_engine cpg_service_engine = {
	.name = "corosync cluster closed process group service v1.01",
	.id = CPG_SERVICE,
	.priority = 1,
	.private_data_size = sizeof (struct cpg_pd),
	.flow_control = CS_LIB_FLOW_CONTROL_REQUIRED,
	.allow_inquorate = CS_LIB_ALLOW_INQUORATE,
	.lib_init_fn = cpg_lib_init_fn,
	.lib_exit_fn = cpg_lib_exit_fn,
	.lib_engine = cpg_lib_engine,
	.lib_engine_count = sizeof (cpg_lib_engine) / sizeof (struct corosync_lib_handler),
	.exec_init_fn = cpg_exec_init_fn,
	.exec_dump_fn = NULL,
	.exec_engine = cpg_exec_engine,
	.exec_engine_count = sizeof (cpg_exec_engine) / sizeof (struct corosync_exec_handler),
	.sync_mode = CS_SYNC_V1_APIV2,
	.sync_init = (sync_init_v1_fn_t)cpg_sync_init_v2,
	.sync_process = cpg_sync_process,
	.sync_activate = cpg_sync_activate,
	.sync_abort = cpg_sync_abort
	};

	/*
	* Dynamic loader definition
	*/
	static struct corosync_service_engine *cpg_get_service_engine_ver0 (void);

	static struct corosync_service_engine_iface_ver0 cpg_service_engine_iface = {
	.corosync_get_service_engine_ver0 = cpg_get_service_engine_ver0
	};

	static struct lcr_iface corosync_cpg_ver0[1] = {
	{
	.name = "corosync_cpg",
	.version = 0,
	.versions_replace = 0,
	.versions_replace_count = 0,
	.dependencies = 0,
	.dependency_count = 0,
	.constructor = NULL,
	.destructor = NULL,
	.interfaces = NULL
	}
	};

	static struct lcr_comp cpg_comp_ver0 = {
	.iface_count = 1,
	.ifaces = corosync_cpg_ver0
	};


	static struct corosync_service_engine *cpg_get_service_engine_ver0 (void)
	{
	return (&cpg_service_engine);
	}

	#ifdef COROSYNC_SOLARIS
	void corosync_lcr_component_register (void);

	void corosync_lcr_component_register (void) {
	#else
	__attribute__ ((constructor)) static void corosync_lcr_component_register (void) {
	#endif
	lcr_interfaces_set (&corosync_cpg_ver0[0], &cpg_service_engine_iface);

	lcr_component_register (&cpg_comp_ver0);
	}

	struct req_exec_cpg_procjoin {
	struct qb_ipc_request_header header __attribute__((aligned(8)));
	mar_cpg_name_t group_name __attribute__((aligned(8)));
	mar_uint32_t pid __attribute__((aligned(8)));
	mar_uint32_t reason __attribute__((aligned(8)));
	};

	struct req_exec_cpg_mcast {
	struct qb_ipc_request_header header __attribute__((aligned(8)));
	mar_cpg_name_t group_name __attribute__((aligned(8)));
	mar_uint32_t msglen __attribute__((aligned(8)));
	mar_uint32_t pid __attribute__((aligned(8)));
	mar_message_source_t source __attribute__((aligned(8)));
	mar_uint8_t message[] __attribute__((aligned(8)));
	};

	struct req_exec_cpg_downlist_old {
	struct qb_ipc_request_header header __attribute__((aligned(8)));
	mar_uint32_t left_nodes __attribute__((aligned(8)));
	mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
	};

	struct req_exec_cpg_downlist {
	struct qb_ipc_request_header header __attribute__((aligned(8)));
	/* merge decisions */
	mar_uint32_t old_members __attribute__((aligned(8)));
	/* downlist below */
	mar_uint32_t left_nodes __attribute__((aligned(8)));
	mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
	};

	struct downlist_msg {
	mar_uint32_t sender_nodeid;
	mar_uint32_t old_members __attribute__((aligned(8)));
	mar_uint32_t left_nodes __attribute__((aligned(8)));
	mar_uint32_t nodeids[PROCESSOR_COUNT_MAX] __attribute__((aligned(8)));
	struct list_head list;
	};

	static struct req_exec_cpg_downlist g_req_exec_cpg_downlist;

	static void cpg_sync_init_v2 (
	const unsigned int *trans_list,
	size_t trans_list_entries,
	const unsigned int *member_list,
	size_t member_list_entries,
	const struct memb_ring_id *ring_id)
	{
	int entries;
	int i, j;
	int found;

	my_sync_state = CPGSYNC_DOWNLIST;

	memcpy (my_member_list, member_list, member_list_entries *
	sizeof (unsigned int));
	my_member_list_entries = member_list_entries;

	last_sync_ring_id.nodeid = ring_id->rep.nodeid;
	last_sync_ring_id.seq = ring_id->seq;

	downlist_state = CPG_DOWNLIST_WAITING_FOR_MESSAGES;

	entries = 0;
	/*
	* Determine list of nodeids for downlist message
	*/
	for (i = 0; i < my_old_member_list_entries; i++) {
	found = 0;
	for (j = 0; j < trans_list_entries; j++) {
	if (my_old_member_list[i] == trans_list[j]) {
	found = 1;
	break;
	}
	}
	if (found == 0) {
	g_req_exec_cpg_downlist.nodeids[entries++] =
	my_old_member_list[i];
	}
	}
	g_req_exec_cpg_downlist.left_nodes = entries;
	}

	static int cpg_sync_process (void)
	{
	int res = -1;

	if (my_sync_state == CPGSYNC_DOWNLIST) {
	res = cpg_exec_send_downlist();
	if (res == -1) {
	return (-1);
	}
	my_sync_state = CPGSYNC_JOINLIST;
	}
	if (my_sync_state == CPGSYNC_JOINLIST) {
	res = cpg_exec_send_joinlist();
	}
	return (res);
	}

	static void cpg_sync_activate (void)
	{
	memcpy (my_old_member_list, my_member_list,
	my_member_list_entries * sizeof (unsigned int));
	my_old_member_list_entries = my_member_list_entries;

	if (downlist_state == CPG_DOWNLIST_WAITING_FOR_MESSAGES) {
	downlist_master_choose_and_send ();
	}

	downlist_messages_delete ();
	downlist_state = CPG_DOWNLIST_NONE;

	notify_lib_totem_membership (NULL, my_member_list_entries, my_member_list);
	}

	static void cpg_sync_abort (void)
	{
	downlist_state = CPG_DOWNLIST_NONE;
	downlist_messages_delete ();
	}

	static int notify_lib_totem_membership (
	void *conn,
	int member_list_entries,
	const unsigned int *member_list)
	{
	struct list_head *iter;
	char *buf;
	int size;
	struct res_lib_cpg_totem_confchg_callback *res;

	size = sizeof(struct res_lib_cpg_totem_confchg_callback) +
	sizeof(mar_uint32_t) * (member_list_entries);
	buf = alloca(size);
	if (!buf)
	return CS_ERR_LIBRARY;

	res = (struct res_lib_cpg_totem_confchg_callback *)buf;
	res->member_list_entries = member_list_entries;
	res->header.size = size;
	res->header.id = MESSAGE_RES_CPG_TOTEM_CONFCHG_CALLBACK;
	res->header.error = CS_OK;

	memcpy (&res->ring_id, &last_sync_ring_id, sizeof (mar_cpg_ring_id_t));
	memcpy (res->member_list, member_list, res->member_list_entries * sizeof (mar_uint32_t));

	if (conn == NULL) {
	for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
	struct cpg_pd *cpg_pd = list_entry (iter, struct cpg_pd, list);
	api->ipc_dispatch_send (cpg_pd->conn, buf, size);
	}
	} else {
	api->ipc_dispatch_send (conn, buf, size);
	}

	return CS_OK;
	}

	static int notify_lib_joinlist(
	const mar_cpg_name_t *group_name,
	void *conn,
	int joined_list_entries,
	mar_cpg_address_t *joined_list,
	int left_list_entries,
	mar_cpg_address_t *left_list,
	int id)
	{
	int size;
	char *buf;
	struct list_head *iter;
	int count;
	struct res_lib_cpg_confchg_callback *res;
	mar_cpg_address_t *retgi;

	count = 0;

	for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
	struct process_info *pi = list_entry (iter, struct process_info, list);
	if (mar_name_compare (&pi->group, group_name) == 0) {
	int i;
	int founded = 0;

	for (i = 0; i < left_list_entries; i++) {
	if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) {
	founded++;
	}
	}

	if (!founded)
	count++;
	}
	}

	size = sizeof(struct res_lib_cpg_confchg_callback) +
	sizeof(mar_cpg_address_t) * (count + left_list_entries + joined_list_entries);
	buf = alloca(size);
	if (!buf)
	return CS_ERR_LIBRARY;

	res = (struct res_lib_cpg_confchg_callback *)buf;
	res->joined_list_entries = joined_list_entries;
	res->left_list_entries = left_list_entries;
	res->member_list_entries = count;
	retgi = res->member_list;
	res->header.size = size;
	res->header.id = id;
	res->header.error = CS_OK;
	memcpy(&res->group_name, group_name, sizeof(mar_cpg_name_t));

	for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
	struct process_info *pi=list_entry (iter, struct process_info, list);

	if (mar_name_compare (&pi->group, group_name) == 0) {
	int i;
	int founded = 0;

	for (i = 0;i < left_list_entries; i++) {
	if (left_list[i].nodeid == pi->nodeid && left_list[i].pid == pi->pid) {
	founded++;
	}
	}

	if (!founded) {
	retgi->nodeid = pi->nodeid;
	retgi->pid = pi->pid;
	retgi++;
	}
	}
	}

	if (left_list_entries) {
	memcpy (retgi, left_list, left_list_entries * sizeof(mar_cpg_address_t));
	retgi += left_list_entries;
	}

	if (joined_list_entries) {
	memcpy (retgi, joined_list, joined_list_entries * sizeof(mar_cpg_address_t));
	retgi += joined_list_entries;
	}

	if (conn) {
	api->ipc_dispatch_send (conn, buf, size);
	} else {
	for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
	struct cpg_pd *cpd = list_entry (iter, struct cpg_pd, list);
	if (mar_name_compare (&cpd->group_name, group_name) == 0) {
	assert (left_list_entries <= 1);
	assert (joined_list_entries <= 1);
	if (joined_list_entries) {
	if (joined_list[0].pid == cpd->pid &&
	joined_list[0].nodeid == api->totem_nodeid_get()) {
	cpd->cpd_state = CPD_STATE_JOIN_COMPLETED;
	}
	}
	if (cpd->cpd_state == CPD_STATE_JOIN_COMPLETED \|\|
	cpd->cpd_state == CPD_STATE_LEAVE_STARTED) {

	api->ipc_dispatch_send (cpd->conn, buf, size);
	}
	if (left_list_entries) {
	if (left_list[0].pid == cpd->pid &&
	left_list[0].nodeid == api->totem_nodeid_get()) {

	cpd->pid = 0;
	memset (&cpd->group_name, 0, sizeof(cpd->group_name));
	cpd->cpd_state = CPD_STATE_UNJOINED;
	}
	}
	}
	}
	}


	/*
	* Traverse thru cpds and send totem membership for cpd, where it is not send yet
	*/
	for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
	struct cpg_pd *cpd = list_entry (iter, struct cpg_pd, list);

	if ((cpd->flags & CPG_MODEL_V1_DELIVER_INITIAL_TOTEM_CONF) && (cpd->initial_totem_conf_sent == 0)) {
	cpd->initial_totem_conf_sent = 1;

	notify_lib_totem_membership (cpd->conn, my_old_member_list_entries, my_old_member_list);
	}
	}

	return CS_OK;
	}

	static void downlist_log(const char msg, struct downlist_msg dl)
	{
	log_printf (LOG_DEBUG,
	"%s: sender %s; members(old:%d left:%d)",
	msg,
	api->totem_ifaces_print(dl->sender_nodeid),
	dl->old_members,
	dl->left_nodes);
	}

	static struct downlist_msg* downlist_master_choose (void)
	{
	struct downlist_msg *cmp;
	struct downlist_msg *best = NULL;
	struct list_head *iter;
	uint32_t cmp_members;
	uint32_t best_members;

	for (iter = downlist_messages_head.next;
	iter != &downlist_messages_head;
	iter = iter->next) {

	cmp = list_entry(iter, struct downlist_msg, list);
	downlist_log("comparing", cmp);
	if (best == NULL) {
	best = cmp;
	continue;
	}
	best_members = best->old_members - best->left_nodes;
	cmp_members = cmp->old_members - cmp->left_nodes;

	if (cmp_members < best_members) {
	continue;
	}
	else if (cmp_members > best_members) {
	best = cmp;
	}
	else if (cmp->sender_nodeid < best->sender_nodeid) {
	best = cmp;
	}

	}
	return best;
	}

	static void downlist_master_choose_and_send (void)
	{
	struct downlist_msg *stored_msg;
	struct list_head *iter;
	mar_cpg_address_t left_list;
	int i;

	downlist_state = CPG_DOWNLIST_APPLYING;

	stored_msg = downlist_master_choose ();
	if (!stored_msg) {
	log_printf (LOGSYS_LEVEL_DEBUG, "NO chosen downlist");
	return;
	}
	downlist_log("chosen downlist", stored_msg);

	/* send events */
	for (iter = process_info_list_head.next; iter != &process_info_list_head; ) {
	struct process_info *pi = list_entry(iter, struct process_info, list);
	iter = iter->next;

	for (i = 0; i < stored_msg->left_nodes; i++) {
	if (pi->nodeid == stored_msg->nodeids[i]) {
	left_list.nodeid = pi->nodeid;
	left_list.pid = pi->pid;
	left_list.reason = CONFCHG_CPG_REASON_NODEDOWN;

	notify_lib_joinlist(&pi->group, NULL,
	0, NULL,
	1, &left_list,
	MESSAGE_RES_CPG_CONFCHG_CALLBACK);
	list_del (&pi->list);
	free (pi);
	break;
	}
	}
	}
	}

	static void downlist_messages_delete (void)
	{
	struct downlist_msg *stored_msg;
	struct list_head iter, iter_next;

	for (iter = downlist_messages_head.next;
	iter != &downlist_messages_head;
	iter = iter_next) {

	iter_next = iter->next;

	stored_msg = list_entry(iter, struct downlist_msg, list);
	list_del (&stored_msg->list);
	free (stored_msg);
	}
	}


	static int cpg_exec_init_fn (struct corosync_api_v1 *corosync_api)
	{
	#ifdef COROSYNC_SOLARIS
	logsys_subsys_init();
	#endif
	list_init (&downlist_messages_head);
	api = corosync_api;
	return (0);
	}

	static void cpg_iteration_instance_finalize (struct cpg_iteration_instance *cpg_iteration_instance)
	{
	struct list_head iter, iter_next;
	struct process_info *pi;

	for (iter = cpg_iteration_instance->items_list_head.next;
	iter != &cpg_iteration_instance->items_list_head;
	iter = iter_next) {

	iter_next = iter->next;

	pi = list_entry (iter, struct process_info, list);
	list_del (&pi->list);
	free (pi);
	}

	list_del (&cpg_iteration_instance->list);
	hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);
	}

	static void cpg_pd_finalize (struct cpg_pd *cpd)
	{
	struct list_head iter, iter_next;
	struct cpg_iteration_instance *cpii;

	zcb_all_free(cpd);
	for (iter = cpd->iteration_instance_list_head.next;
	iter != &cpd->iteration_instance_list_head;
	iter = iter_next) {

	iter_next = iter->next;

	cpii = list_entry (iter, struct cpg_iteration_instance, list);

	cpg_iteration_instance_finalize (cpii);
	}

	list_del (&cpd->list);
	}

	static int cpg_lib_exit_fn (void *conn)
	{
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);

	log_printf(LOGSYS_LEVEL_DEBUG, "exit_fn for conn=%p\n", conn);

	if (cpd->group_name.length > 0) {
	cpg_node_joinleave_send (cpd->pid, &cpd->group_name,
	MESSAGE_REQ_EXEC_CPG_PROCLEAVE, CONFCHG_CPG_REASON_PROCDOWN);
	}

	cpg_pd_finalize (cpd);

	api->ipc_refcnt_dec (conn);
	return (0);
	}

	static int cpg_node_joinleave_send (unsigned int pid, const mar_cpg_name_t *group_name, int fn, int reason)
	{
	struct req_exec_cpg_procjoin req_exec_cpg_procjoin;
	struct iovec req_exec_cpg_iovec;
	int result;

	memcpy(&req_exec_cpg_procjoin.group_name, group_name, sizeof(mar_cpg_name_t));
	req_exec_cpg_procjoin.pid = pid;
	req_exec_cpg_procjoin.reason = reason;

	req_exec_cpg_procjoin.header.size = sizeof(req_exec_cpg_procjoin);
	req_exec_cpg_procjoin.header.id = SERVICE_ID_MAKE(CPG_SERVICE, fn);

	req_exec_cpg_iovec.iov_base = (char *)&req_exec_cpg_procjoin;
	req_exec_cpg_iovec.iov_len = sizeof(req_exec_cpg_procjoin);

	result = api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED);

	return (result);
	}

	/* Can byteswap join & leave messages */
	static void exec_cpg_procjoin_endian_convert (void *msg)
	{
	struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = msg;

	req_exec_cpg_procjoin->pid = swab32(req_exec_cpg_procjoin->pid);
	swab_mar_cpg_name_t (&req_exec_cpg_procjoin->group_name);
	req_exec_cpg_procjoin->reason = swab32(req_exec_cpg_procjoin->reason);
	}

	static void exec_cpg_joinlist_endian_convert (void *msg_v)
	{
	char *msg = msg_v;
	struct qb_ipc_response_header res = (struct qb_ipc_response_header )msg;
	struct join_list_entry jle = (struct join_list_entry )(msg + sizeof(struct qb_ipc_response_header));

	swab_mar_int32_t (&res->size);

	while ((const char*)jle < msg + res->size) {
	jle->pid = swab32(jle->pid);
	swab_mar_cpg_name_t (&jle->group_name);
	jle++;
	}
	}

	static void exec_cpg_downlist_endian_convert_old (void *msg)
	{
	}

	static void exec_cpg_downlist_endian_convert (void *msg)
	{
	struct req_exec_cpg_downlist *req_exec_cpg_downlist = msg;
	unsigned int i;

	req_exec_cpg_downlist->left_nodes = swab32(req_exec_cpg_downlist->left_nodes);
	req_exec_cpg_downlist->old_members = swab32(req_exec_cpg_downlist->old_members);

	for (i = 0; i < req_exec_cpg_downlist->left_nodes; i++) {
	req_exec_cpg_downlist->nodeids[i] = swab32(req_exec_cpg_downlist->nodeids[i]);
	}
	}


	static void exec_cpg_mcast_endian_convert (void *msg)
	{
	struct req_exec_cpg_mcast *req_exec_cpg_mcast = msg;

	swab_coroipc_request_header_t (&req_exec_cpg_mcast->header);
	swab_mar_cpg_name_t (&req_exec_cpg_mcast->group_name);
	req_exec_cpg_mcast->pid = swab32(req_exec_cpg_mcast->pid);
	req_exec_cpg_mcast->msglen = swab32(req_exec_cpg_mcast->msglen);
	swab_mar_message_source_t (&req_exec_cpg_mcast->source);
	}

	static struct process_info process_info_find(const mar_cpg_name_t group_name, uint32_t pid, unsigned int nodeid) {
	struct list_head *iter;

	for (iter = process_info_list_head.next; iter != &process_info_list_head; ) {
	struct process_info *pi = list_entry (iter, struct process_info, list);
	iter = iter->next;

	if (pi->pid == pid && pi->nodeid == nodeid &&
	mar_name_compare (&pi->group, group_name) == 0) {
	return pi;
	}
	}

	return NULL;
	}

	static void do_proc_join(
	const mar_cpg_name_t *name,
	uint32_t pid,
	unsigned int nodeid,
	int reason)
	{
	struct process_info *pi;
	struct process_info *pi_entry;
	mar_cpg_address_t notify_info;
	struct list_head *list;
	struct list_head *list_to_add = NULL;

	if (process_info_find (name, pid, nodeid) != NULL) {
	return ;
	}
	pi = malloc (sizeof (struct process_info));
	if (!pi) {
	log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");
	return;
	}
	pi->nodeid = nodeid;
	pi->pid = pid;
	memcpy(&pi->group, name, sizeof(*name));
	list_init(&pi->list);

	/*
	* Insert new process in sorted order so synchronization works properly
	*/
	list_to_add = &process_info_list_head;
	for (list = process_info_list_head.next; list != &process_info_list_head; list = list->next) {

	pi_entry = list_entry(list, struct process_info, list);
	if (pi_entry->nodeid > pi->nodeid \|\|
	(pi_entry->nodeid == pi->nodeid && pi_entry->pid > pi->pid)) {

	break;
	}
	list_to_add = list;
	}
	list_add (&pi->list, list_to_add);

	notify_info.pid = pi->pid;
	notify_info.nodeid = nodeid;
	notify_info.reason = reason;

	notify_lib_joinlist(&pi->group, NULL,
	1, &notify_info,
	0, NULL,
	MESSAGE_RES_CPG_CONFCHG_CALLBACK);
	}

	static void message_handler_req_exec_cpg_downlist_old (
	const void *message,
	unsigned int nodeid)
	{
	log_printf (LOGSYS_LEVEL_WARNING, "downlist OLD from node %d",
	nodeid);
	}

	static void message_handler_req_exec_cpg_downlist(
	const void *message,
	unsigned int nodeid)
	{
	const struct req_exec_cpg_downlist *req_exec_cpg_downlist = message;
	int i;
	struct list_head *iter;
	struct downlist_msg *stored_msg;
	int found;

	if (downlist_state != CPG_DOWNLIST_WAITING_FOR_MESSAGES) {
	log_printf (LOGSYS_LEVEL_WARNING, "downlist left_list: %d received in state %d",
	req_exec_cpg_downlist->left_nodes, downlist_state);
	return;
	}

	stored_msg = malloc (sizeof (struct downlist_msg));
	stored_msg->sender_nodeid = nodeid;
	stored_msg->old_members = req_exec_cpg_downlist->old_members;
	stored_msg->left_nodes = req_exec_cpg_downlist->left_nodes;
	memcpy (stored_msg->nodeids, req_exec_cpg_downlist->nodeids,
	req_exec_cpg_downlist->left_nodes * sizeof (mar_uint32_t));
	list_init (&stored_msg->list);
	list_add (&stored_msg->list, &downlist_messages_head);

	for (i = 0; i < my_member_list_entries; i++) {
	found = 0;
	for (iter = downlist_messages_head.next;
	iter != &downlist_messages_head;
	iter = iter->next) {

	stored_msg = list_entry(iter, struct downlist_msg, list);
	if (my_member_list[i] == stored_msg->sender_nodeid) {
	found = 1;
	}
	}
	if (!found) {
	return;
	}
	}

	downlist_master_choose_and_send ();
	}


	static void message_handler_req_exec_cpg_procjoin (
	const void *message,
	unsigned int nodeid)
	{
	const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;

	log_printf(LOGSYS_LEVEL_DEBUG, "got procjoin message from cluster node %d\n", nodeid);

	do_proc_join (&req_exec_cpg_procjoin->group_name,
	req_exec_cpg_procjoin->pid, nodeid,
	CONFCHG_CPG_REASON_JOIN);
	}

	static void message_handler_req_exec_cpg_procleave (
	const void *message,
	unsigned int nodeid)
	{
	const struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = message;
	struct process_info *pi;
	struct list_head *iter;
	mar_cpg_address_t notify_info;

	log_printf(LOGSYS_LEVEL_DEBUG, "got procleave message from cluster node %d\n", nodeid);

	notify_info.pid = req_exec_cpg_procjoin->pid;
	notify_info.nodeid = nodeid;
	notify_info.reason = req_exec_cpg_procjoin->reason;

	notify_lib_joinlist(&req_exec_cpg_procjoin->group_name, NULL,
	0, NULL,
	1, &notify_info,
	MESSAGE_RES_CPG_CONFCHG_CALLBACK);

	for (iter = process_info_list_head.next; iter != &process_info_list_head; ) {
	pi = list_entry(iter, struct process_info, list);
	iter = iter->next;

	if (pi->pid == req_exec_cpg_procjoin->pid && pi->nodeid == nodeid &&
	mar_name_compare (&pi->group, &req_exec_cpg_procjoin->group_name)==0) {
	list_del (&pi->list);
	free (pi);
	}
	}
	}


	/* Got a proclist from another node */
	static void message_handler_req_exec_cpg_joinlist (
	const void *message_v,
	unsigned int nodeid)
	{
	const char *message = message_v;
	const struct qb_ipc_response_header res = (const struct qb_ipc_response_header )message;
	const struct join_list_entry jle = (const struct join_list_entry )(message + sizeof(struct qb_ipc_response_header));

	log_printf(LOGSYS_LEVEL_DEBUG, "got joinlist message from node %x\n",
	nodeid);

	/* Ignore our own messages */
	if (nodeid == api->totem_nodeid_get()) {
	return;
	}

	while ((const char*)jle < message + res->size) {
	do_proc_join (&jle->group_name, jle->pid, nodeid,
	CONFCHG_CPG_REASON_NODEUP);
	jle++;
	}
	}

	static void message_handler_req_exec_cpg_mcast (
	const void *message,
	unsigned int nodeid)
	{
	const struct req_exec_cpg_mcast *req_exec_cpg_mcast = message;
	struct res_lib_cpg_deliver_callback res_lib_cpg_mcast;
	int msglen = req_exec_cpg_mcast->msglen;
	struct list_head iter, pi_iter;
	struct cpg_pd *cpd;
	struct iovec iovec[2];
	int known_node = 0;

	res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_DELIVER_CALLBACK;
	res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast) + msglen;
	res_lib_cpg_mcast.msglen = msglen;
	res_lib_cpg_mcast.pid = req_exec_cpg_mcast->pid;
	res_lib_cpg_mcast.nodeid = nodeid;

	memcpy(&res_lib_cpg_mcast.group_name, &req_exec_cpg_mcast->group_name,
	sizeof(mar_cpg_name_t));
	iovec[0].iov_base = (void *)&res_lib_cpg_mcast;
	iovec[0].iov_len = sizeof (res_lib_cpg_mcast);

	iovec[1].iov_base = (char)message+sizeof(req_exec_cpg_mcast);
	iovec[1].iov_len = msglen;

	for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; ) {
	cpd = list_entry(iter, struct cpg_pd, list);
	iter = iter->next;

	if ((cpd->cpd_state == CPD_STATE_LEAVE_STARTED \|\| cpd->cpd_state == CPD_STATE_JOIN_COMPLETED)
	&& (mar_name_compare (&cpd->group_name, &req_exec_cpg_mcast->group_name) == 0)) {

	if (!known_node) {
	/* Try to find, if we know the node */
	for (pi_iter = process_info_list_head.next;
	pi_iter != &process_info_list_head; pi_iter = pi_iter->next) {

	struct process_info *pi = list_entry (pi_iter, struct process_info, list);

	if (pi->nodeid == nodeid &&
	mar_name_compare (&pi->group, &req_exec_cpg_mcast->group_name) == 0) {
	known_node = 1;
	break;
	}
	}
	}

	if (!known_node) {
	log_printf(LOGSYS_LEVEL_WARNING, "Unknown node -> we will not deliver message");
	return ;
	}

	api->ipc_dispatch_iov_send (cpd->conn, iovec, 2);
	}
	}
	}


	static int cpg_exec_send_downlist(void)
	{
	struct iovec iov;

	g_req_exec_cpg_downlist.header.id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_DOWNLIST);
	g_req_exec_cpg_downlist.header.size = sizeof(struct req_exec_cpg_downlist);

	g_req_exec_cpg_downlist.old_members = my_old_member_list_entries;

	iov.iov_base = (void *)&g_req_exec_cpg_downlist;
	iov.iov_len = g_req_exec_cpg_downlist.header.size;

	return (api->totem_mcast (&iov, 1, TOTEM_AGREED));
	}

	static int cpg_exec_send_joinlist(void)
	{
	int count = 0;
	struct list_head *iter;
	struct qb_ipc_response_header *res;
	char *buf;
	struct join_list_entry *jle;
	struct iovec req_exec_cpg_iovec;

	for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
	struct process_info *pi = list_entry (iter, struct process_info, list);

	if (pi->nodeid == api->totem_nodeid_get ()) {
	count++;
	}
	}

	/* Nothing to send */
	if (!count)
	return 0;

	buf = alloca(sizeof(struct qb_ipc_response_header) + sizeof(struct join_list_entry) * count);
	if (!buf) {
	log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate joinlist buffer");
	return -1;
	}

	jle = (struct join_list_entry *)(buf + sizeof(struct qb_ipc_response_header));
	res = (struct qb_ipc_response_header *)buf;

	for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
	struct process_info *pi = list_entry (iter, struct process_info, list);

	if (pi->nodeid == api->totem_nodeid_get ()) {
	memcpy (&jle->group_name, &pi->group, sizeof (mar_cpg_name_t));
	jle->pid = pi->pid;
	jle++;
	}
	}

	res->id = SERVICE_ID_MAKE(CPG_SERVICE, MESSAGE_REQ_EXEC_CPG_JOINLIST);
	res->size = sizeof(struct qb_ipc_response_header)+sizeof(struct join_list_entry) * count;

	req_exec_cpg_iovec.iov_base = buf;
	req_exec_cpg_iovec.iov_len = res->size;

	return (api->totem_mcast (&req_exec_cpg_iovec, 1, TOTEM_AGREED));
	}

	static int cpg_lib_init_fn (void *conn)
	{
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);
	memset (cpd, 0, sizeof(struct cpg_pd));
	cpd->conn = conn;
	list_add (&cpd->list, &cpg_pd_list_head);

	list_init (&cpd->iteration_instance_list_head);
	list_init (&cpd->zcb_mapped_list_head);

	api->ipc_refcnt_inc (conn);
	log_printf(LOGSYS_LEVEL_DEBUG, "lib_init_fn: conn=%p, cpd=%p\n", conn, cpd);
	return (0);
	}

	/* Join message from the library */
	static void message_handler_req_lib_cpg_join (void conn, const void message)
	{
	const struct req_lib_cpg_join *req_lib_cpg_join = message;
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);
	struct res_lib_cpg_join res_lib_cpg_join;
	cs_error_t error = CS_OK;
	struct list_head *iter;

	/* Test, if we don't have same pid and group name joined */
	for (iter = cpg_pd_list_head.next; iter != &cpg_pd_list_head; iter = iter->next) {
	struct cpg_pd *cpd_item = list_entry (iter, struct cpg_pd, list);

	if (cpd_item->pid == req_lib_cpg_join->pid &&
	mar_name_compare(&req_lib_cpg_join->group_name, &cpd_item->group_name) == 0) {

	/* We have same pid and group name joined -> return error */
	error = CS_ERR_EXIST;
	goto response_send;
	}
	}

	/*
	* Same check must be done in process info list, because there may be not yet delivered
	* leave of client.
	*/
	for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
	struct process_info *pi = list_entry (iter, struct process_info, list);

	if (pi->nodeid == api->totem_nodeid_get () && pi->pid == req_lib_cpg_join->pid &&
	mar_name_compare(&req_lib_cpg_join->group_name, &pi->group) == 0) {
	/* We have same pid and group name joined -> return error */
	error = CS_ERR_TRY_AGAIN;
	goto response_send;
	}
	}

	switch (cpd->cpd_state) {
	case CPD_STATE_UNJOINED:
	error = CS_OK;
	cpd->cpd_state = CPD_STATE_JOIN_STARTED;
	cpd->pid = req_lib_cpg_join->pid;
	cpd->flags = req_lib_cpg_join->flags;
	memcpy (&cpd->group_name, &req_lib_cpg_join->group_name,
	sizeof (cpd->group_name));

	cpg_node_joinleave_send (req_lib_cpg_join->pid,
	&req_lib_cpg_join->group_name,
	MESSAGE_REQ_EXEC_CPG_PROCJOIN, CONFCHG_CPG_REASON_JOIN);
	break;
	case CPD_STATE_LEAVE_STARTED:
	error = CS_ERR_BUSY;
	break;
	case CPD_STATE_JOIN_STARTED:
	error = CS_ERR_EXIST;
	break;
	case CPD_STATE_JOIN_COMPLETED:
	error = CS_ERR_EXIST;
	break;
	}

	response_send:
	res_lib_cpg_join.header.size = sizeof(res_lib_cpg_join);
	res_lib_cpg_join.header.id = MESSAGE_RES_CPG_JOIN;
	res_lib_cpg_join.header.error = error;
	api->ipc_response_send (conn, &res_lib_cpg_join, sizeof(res_lib_cpg_join));
	}

	/* Leave message from the library */
	static void message_handler_req_lib_cpg_leave (void conn, const void message)
	{
	struct res_lib_cpg_leave res_lib_cpg_leave;
	cs_error_t error = CS_OK;
	struct req_lib_cpg_leave req_lib_cpg_leave = (struct req_lib_cpg_leave )message;
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);

	log_printf(LOGSYS_LEVEL_DEBUG, "got leave request on %p\n", conn);

	switch (cpd->cpd_state) {
	case CPD_STATE_UNJOINED:
	error = CS_ERR_NOT_EXIST;
	break;
	case CPD_STATE_LEAVE_STARTED:
	error = CS_ERR_NOT_EXIST;
	break;
	case CPD_STATE_JOIN_STARTED:
	error = CS_ERR_BUSY;
	break;
	case CPD_STATE_JOIN_COMPLETED:
	error = CS_OK;
	cpd->cpd_state = CPD_STATE_LEAVE_STARTED;
	cpg_node_joinleave_send (req_lib_cpg_leave->pid,
	&req_lib_cpg_leave->group_name,
	MESSAGE_REQ_EXEC_CPG_PROCLEAVE,
	CONFCHG_CPG_REASON_LEAVE);
	break;
	}

	/* send return */
	res_lib_cpg_leave.header.size = sizeof(res_lib_cpg_leave);
	res_lib_cpg_leave.header.id = MESSAGE_RES_CPG_LEAVE;
	res_lib_cpg_leave.header.error = error;
	api->ipc_response_send(conn, &res_lib_cpg_leave, sizeof(res_lib_cpg_leave));
	}

	/* Finalize message from library */
	static void message_handler_req_lib_cpg_finalize (
	void *conn,
	const void *message)
	{
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);
	struct res_lib_cpg_finalize res_lib_cpg_finalize;
	cs_error_t error = CS_OK;

	log_printf (LOGSYS_LEVEL_DEBUG, "cpg finalize for conn=%p\n", conn);

	/*
	* We will just remove cpd from list. After this call, connection will be
	* closed on lib side, and cpg_lib_exit_fn will be called
	*/
	list_del (&cpd->list);
	list_init (&cpd->list);

	res_lib_cpg_finalize.header.size = sizeof (res_lib_cpg_finalize);
	res_lib_cpg_finalize.header.id = MESSAGE_RES_CPG_FINALIZE;
	res_lib_cpg_finalize.header.error = error;

	api->ipc_response_send (conn, &res_lib_cpg_finalize,
	sizeof (res_lib_cpg_finalize));
	}

	static int
	memory_map (
	const char *path,
	size_t bytes,
	void **buf)
	{
	int32_t fd;
	void *addr_orig;
	void *addr;
	int32_t res;

	fd = open (path, O_RDWR, 0600);

	unlink (path);

	if (fd == -1) {
	return (-1);
	}

	res = ftruncate (fd, bytes);
	if (res == -1) {
	goto error_close_unlink;
	}

	addr_orig = mmap (NULL, bytes, PROT_NONE,
	MAP_ANONYMOUS \| MAP_PRIVATE, -1, 0);

	if (addr_orig == MAP_FAILED) {
	goto error_close_unlink;
	}

	addr = mmap (addr_orig, bytes, PROT_READ \| PROT_WRITE,
	MAP_FIXED \| MAP_SHARED, fd, 0);

	if (addr != addr_orig) {
	munmap(addr_orig, bytes);
	goto error_close_unlink;
	}
	#ifdef COROSYNC_BSD
	madvise(addr, bytes, MADV_NOSYNC);
	#endif

	res = close (fd);
	if (res) {
	return (-1);
	}
	*buf = addr_orig;
	return (0);

	error_close_unlink:
	close (fd);
	unlink(path);
	return -1;
	}

	static inline int zcb_alloc (
	struct cpg_pd *cpd,
	const char *path_to_file,
	size_t size,
	void **addr)
	{
	struct zcb_mapped *zcb_mapped;
	unsigned int res;

	zcb_mapped = malloc (sizeof (struct zcb_mapped));
	if (zcb_mapped == NULL) {
	return (-1);
	}

	res = memory_map (
	path_to_file,
	size,
	addr);
	if (res == -1) {
	free (zcb_mapped);
	return (-1);
	}

	list_init (&zcb_mapped->list);
	zcb_mapped->addr = *addr;
	zcb_mapped->size = size;
	list_add_tail (&zcb_mapped->list, &cpd->zcb_mapped_list_head);
	return (0);
	}


	static inline int zcb_free (struct zcb_mapped *zcb_mapped)
	{
	unsigned int res;

	res = munmap (zcb_mapped->addr, zcb_mapped->size);
	list_del (&zcb_mapped->list);
	free (zcb_mapped);
	return (res);
	}

	static inline int zcb_by_addr_free (struct cpg_pd cpd, void addr)
	{
	struct list_head *list;
	struct zcb_mapped *zcb_mapped;
	unsigned int res = 0;

	for (list = cpd->zcb_mapped_list_head.next;
	list != &cpd->zcb_mapped_list_head; list = list->next) {

	zcb_mapped = list_entry (list, struct zcb_mapped, list);

	if (zcb_mapped->addr == addr) {
	res = zcb_free (zcb_mapped);
	break;
	}

	}
	return (res);
	}

	static inline int zcb_all_free (
	struct cpg_pd *cpd)
	{
	struct list_head *list;
	struct zcb_mapped *zcb_mapped;

	for (list = cpd->zcb_mapped_list_head.next;
	list != &cpd->zcb_mapped_list_head;) {

	zcb_mapped = list_entry (list, struct zcb_mapped, list);

	list = list->next;

	zcb_free (zcb_mapped);
	}
	return (0);
	}

	union u {
	uint64_t server_addr;
	void *server_ptr;
	};

	static uint64_t void2serveraddr (void *server_ptr)
	{
	union u u;

	u.server_ptr = server_ptr;
	return (u.server_addr);
	}

	static void *serveraddr2void (uint64_t server_addr)
	{
	union u u;

	u.server_addr = server_addr;
	return (u.server_ptr);
	};

	static void message_handler_req_lib_cpg_zc_alloc (
	void *conn,
	const void *message)
	{
	mar_req_coroipcc_zc_alloc_t hdr = (mar_req_coroipcc_zc_alloc_t )message;
	struct qb_ipc_response_header res_header;
	void *addr = NULL;
	struct coroipcs_zc_header *zc_header;
	unsigned int res;
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);

	log_printf(LOGSYS_LEVEL_DEBUG, "path: %s", hdr->path_to_file);

	res = zcb_alloc (cpd, hdr->path_to_file, hdr->map_size,
	&addr);
	assert(res == 0);

	zc_header = (struct coroipcs_zc_header *)addr;
	zc_header->server_address = void2serveraddr(addr);

	res_header.size = sizeof (struct qb_ipc_response_header);
	res_header.id = 0;
	api->ipc_response_send (conn,
	&res_header,
	res_header.size);
	}

	static void message_handler_req_lib_cpg_zc_free (
	void *conn,
	const void *message)
	{
	mar_req_coroipcc_zc_free_t hdr = (mar_req_coroipcc_zc_free_t )message;
	struct qb_ipc_response_header res_header;
	void *addr = NULL;
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);

	log_printf(LOGSYS_LEVEL_DEBUG, " free'ing");

	addr = serveraddr2void (hdr->server_address);

	zcb_by_addr_free (cpd, addr);

	res_header.size = sizeof (struct qb_ipc_response_header);
	res_header.id = 0;
	api->ipc_response_send (
	conn, &res_header,
	res_header.size);
	}

	/* Mcast message from the library */
	static void message_handler_req_lib_cpg_mcast (void conn, const void message)
	{
	const struct req_lib_cpg_mcast *req_lib_cpg_mcast = message;
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);
	mar_cpg_name_t group_name = cpd->group_name;

	struct iovec req_exec_cpg_iovec[2];
	struct req_exec_cpg_mcast req_exec_cpg_mcast;
	int msglen = req_lib_cpg_mcast->msglen;
	int result;
	cs_error_t error = CS_ERR_NOT_EXIST;

	log_printf(LOGSYS_LEVEL_DEBUG, "got mcast request on %p\n", conn);

	switch (cpd->cpd_state) {
	case CPD_STATE_UNJOINED:
	error = CS_ERR_NOT_EXIST;
	break;
	case CPD_STATE_LEAVE_STARTED:
	error = CS_ERR_NOT_EXIST;
	break;
	case CPD_STATE_JOIN_STARTED:
	error = CS_OK;
	break;
	case CPD_STATE_JOIN_COMPLETED:
	error = CS_OK;
	break;
	}

	if (error == CS_OK) {
	req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen;
	req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
	MESSAGE_REQ_EXEC_CPG_MCAST);
	req_exec_cpg_mcast.pid = cpd->pid;
	req_exec_cpg_mcast.msglen = msglen;
	api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
	memcpy(&req_exec_cpg_mcast.group_name, &group_name,
	sizeof(mar_cpg_name_t));

	req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
	req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
	req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message;
	req_exec_cpg_iovec[1].iov_len = msglen;

	result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
	assert(result == 0);
	} else {
	log_printf(LOGSYS_LEVEL_ERROR, "*** %p can't mcast to group %s state:%d, error:%d\n",
	conn, group_name.value, cpd->cpd_state, error);
	}
	}

	static void message_handler_req_lib_cpg_zc_execute (
	void *conn,
	const void *message)
	{
	mar_req_coroipcc_zc_execute_t hdr = (mar_req_coroipcc_zc_execute_t )message;
	struct qb_ipc_request_header *header;
	struct res_lib_cpg_mcast res_lib_cpg_mcast;
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);
	struct iovec req_exec_cpg_iovec[2];
	struct req_exec_cpg_mcast req_exec_cpg_mcast;
	struct req_lib_cpg_mcast *req_lib_cpg_mcast;
	int result;
	cs_error_t error = CS_ERR_NOT_EXIST;

	log_printf(LOGSYS_LEVEL_DEBUG, "got ZC mcast request on %p\n", conn);

	header = (struct qb_ipc_request_header )(((char )serveraddr2void(hdr->server_address) + sizeof (struct coroipcs_zc_header)));
	req_lib_cpg_mcast = (struct req_lib_cpg_mcast *)header;

	switch (cpd->cpd_state) {
	case CPD_STATE_UNJOINED:
	error = CS_ERR_NOT_EXIST;
	break;
	case CPD_STATE_LEAVE_STARTED:
	error = CS_ERR_NOT_EXIST;
	break;
	case CPD_STATE_JOIN_STARTED:
	error = CS_OK;
	break;
	case CPD_STATE_JOIN_COMPLETED:
	error = CS_OK;
	break;
	}

	res_lib_cpg_mcast.header.size = sizeof(res_lib_cpg_mcast);
	res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_MCAST;
	if (error == CS_OK) {
	req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + req_lib_cpg_mcast->msglen;
	req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
	MESSAGE_REQ_EXEC_CPG_MCAST);
	req_exec_cpg_mcast.pid = cpd->pid;
	req_exec_cpg_mcast.msglen = req_lib_cpg_mcast->msglen;
	api->ipc_source_set (&req_exec_cpg_mcast.source, conn);
	memcpy(&req_exec_cpg_mcast.group_name, &cpd->group_name,
	sizeof(mar_cpg_name_t));

	req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
	req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
	req_exec_cpg_iovec[1].iov_base = (char *)header + sizeof(struct req_lib_cpg_mcast);
	req_exec_cpg_iovec[1].iov_len = req_exec_cpg_mcast.msglen;

	result = api->totem_mcast (req_exec_cpg_iovec, 2, TOTEM_AGREED);
	if (result == 0) {
	res_lib_cpg_mcast.header.error = CS_OK;
	} else {
	res_lib_cpg_mcast.header.error = CS_ERR_TRY_AGAIN;
	}
	} else {
	res_lib_cpg_mcast.header.error = error;
	}

	api->ipc_response_send (conn, &res_lib_cpg_mcast,
	sizeof (res_lib_cpg_mcast));

	}

	static void message_handler_req_lib_cpg_membership (void *conn,
	const void *message)
	{
	struct req_lib_cpg_membership_get *req_lib_cpg_membership_get =
	(struct req_lib_cpg_membership_get *)message;
	struct res_lib_cpg_membership_get res_lib_cpg_membership_get;
	struct list_head *iter;
	int member_count = 0;

	res_lib_cpg_membership_get.header.id = MESSAGE_RES_CPG_MEMBERSHIP;
	res_lib_cpg_membership_get.header.error = CS_OK;
	res_lib_cpg_membership_get.header.size =
	sizeof (struct req_lib_cpg_membership_get);

	for (iter = process_info_list_head.next;
	iter != &process_info_list_head; iter = iter->next) {

	struct process_info *pi = list_entry (iter, struct process_info, list);
	if (mar_name_compare (&pi->group, &req_lib_cpg_membership_get->group_name) == 0) {
	res_lib_cpg_membership_get.member_list[member_count].nodeid = pi->nodeid;
	res_lib_cpg_membership_get.member_list[member_count].pid = pi->pid;
	member_count += 1;
	}
	}
	res_lib_cpg_membership_get.member_count = member_count;

	api->ipc_response_send (conn, &res_lib_cpg_membership_get,
	sizeof (res_lib_cpg_membership_get));
	}

	static void message_handler_req_lib_cpg_local_get (void *conn,
	const void *message)
	{
	struct res_lib_cpg_local_get res_lib_cpg_local_get;

	res_lib_cpg_local_get.header.size = sizeof (res_lib_cpg_local_get);
	res_lib_cpg_local_get.header.id = MESSAGE_RES_CPG_LOCAL_GET;
	res_lib_cpg_local_get.header.error = CS_OK;
	res_lib_cpg_local_get.local_nodeid = api->totem_nodeid_get ();

	api->ipc_response_send (conn, &res_lib_cpg_local_get,
	sizeof (res_lib_cpg_local_get));
	}

	static void message_handler_req_lib_cpg_iteration_initialize (
	void *conn,
	const void *message)
	{
	const struct req_lib_cpg_iterationinitialize *req_lib_cpg_iterationinitialize = message;
	struct cpg_pd cpd = (struct cpg_pd )api->ipc_private_data_get (conn);
	hdb_handle_t cpg_iteration_handle = 0;
	struct res_lib_cpg_iterationinitialize res_lib_cpg_iterationinitialize;
	struct list_head iter, iter2;
	struct cpg_iteration_instance *cpg_iteration_instance;
	cs_error_t error = CS_OK;
	int res;

	log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration initialize\n");

	/* Because between calling this function and *next can be some operations which will
	* change list, we must do full copy.
	*/

	/*
	* Create new iteration instance
	*/
	res = hdb_handle_create (&cpg_iteration_handle_t_db, sizeof (struct cpg_iteration_instance),
	&cpg_iteration_handle);

	if (res != 0) {
	error = CS_ERR_NO_MEMORY;
	goto response_send;
	}

	res = hdb_handle_get (&cpg_iteration_handle_t_db, cpg_iteration_handle, (void *)&cpg_iteration_instance);

	if (res != 0) {
	error = CS_ERR_BAD_HANDLE;
	goto error_destroy;
	}

	list_init (&cpg_iteration_instance->items_list_head);
	cpg_iteration_instance->handle = cpg_iteration_handle;

	/*
	* Create copy of process_info list "grouped by" group name
	*/
	for (iter = process_info_list_head.next; iter != &process_info_list_head; iter = iter->next) {
	struct process_info *pi = list_entry (iter, struct process_info, list);
	struct process_info *new_pi;

	if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
	/*
	* Try to find processed group name in our list new list
	*/
	int found = 0;

	for (iter2 = cpg_iteration_instance->items_list_head.next;
	iter2 != &cpg_iteration_instance->items_list_head;
	iter2 = iter2->next) {
	struct process_info *pi2 = list_entry (iter2, struct process_info, list);

	if (mar_name_compare (&pi2->group, &pi->group) == 0) {
	found = 1;
	break;
	}
	}

	if (found) {
	/*
	* We have this name in list -> don't add
	*/
	continue ;
	}
	} else if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_ONE_GROUP) {
	/*
	* Test pi group name with request
	*/
	if (mar_name_compare (&pi->group, &req_lib_cpg_iterationinitialize->group_name) != 0)
	/*
	* Not same -> don't add
	*/
	continue ;
	}

	new_pi = malloc (sizeof (struct process_info));
	if (!new_pi) {
	log_printf(LOGSYS_LEVEL_WARNING, "Unable to allocate process_info struct");

	error = CS_ERR_NO_MEMORY;

	goto error_put_destroy;
	}

	memcpy (new_pi, pi, sizeof (struct process_info));
	list_init (&new_pi->list);

	if (req_lib_cpg_iterationinitialize->iteration_type == CPG_ITERATION_NAME_ONLY) {
	/*
	* pid and nodeid -> undefined
	*/
	new_pi->pid = new_pi->nodeid = 0;
	}

	/*
	* We will return list "grouped" by "group name", so try to find right place to add
	*/
	for (iter2 = cpg_iteration_instance->items_list_head.next;
	iter2 != &cpg_iteration_instance->items_list_head;
	iter2 = iter2->next) {
	struct process_info *pi2 = list_entry (iter2, struct process_info, list);

	if (mar_name_compare (&pi2->group, &pi->group) == 0) {
	break;
	}
	}

	list_add (&new_pi->list, iter2);
	}

	/*
	* Now we have a full "grouped by" copy of process_info list
	*/

	/*
	* Add instance to current cpd list
	*/
	list_init (&cpg_iteration_instance->list);
	list_add (&cpg_iteration_instance->list, &cpd->iteration_instance_list_head);

	cpg_iteration_instance->current_pointer = &cpg_iteration_instance->items_list_head;

	error_put_destroy:
	hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_handle);
	error_destroy:
	if (error != CS_OK) {
	hdb_handle_destroy (&cpg_iteration_handle_t_db, cpg_iteration_handle);
	}

	response_send:
	res_lib_cpg_iterationinitialize.header.size = sizeof (res_lib_cpg_iterationinitialize);
	res_lib_cpg_iterationinitialize.header.id = MESSAGE_RES_CPG_ITERATIONINITIALIZE;
	res_lib_cpg_iterationinitialize.header.error = error;
	res_lib_cpg_iterationinitialize.iteration_handle = cpg_iteration_handle;

	api->ipc_response_send (conn, &res_lib_cpg_iterationinitialize,
	sizeof (res_lib_cpg_iterationinitialize));
	}

	static void message_handler_req_lib_cpg_iteration_next (
	void *conn,
	const void *message)
	{
	const struct req_lib_cpg_iterationnext *req_lib_cpg_iterationnext = message;
	struct res_lib_cpg_iterationnext res_lib_cpg_iterationnext;
	struct cpg_iteration_instance *cpg_iteration_instance;
	cs_error_t error = CS_OK;
	int res;
	struct process_info *pi;

	log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration next\n");

	res = hdb_handle_get (&cpg_iteration_handle_t_db,
	req_lib_cpg_iterationnext->iteration_handle,
	(void *)&cpg_iteration_instance);

	if (res != 0) {
	error = CS_ERR_LIBRARY;
	goto error_exit;
	}

	assert (cpg_iteration_instance);

	cpg_iteration_instance->current_pointer = cpg_iteration_instance->current_pointer->next;

	if (cpg_iteration_instance->current_pointer == &cpg_iteration_instance->items_list_head) {
	error = CS_ERR_NO_SECTIONS;
	goto error_put;
	}

	pi = list_entry (cpg_iteration_instance->current_pointer, struct process_info, list);

	/*
	* Copy iteration data
	*/
	res_lib_cpg_iterationnext.description.nodeid = pi->nodeid;
	res_lib_cpg_iterationnext.description.pid = pi->pid;
	memcpy (&res_lib_cpg_iterationnext.description.group,
	&pi->group,
	sizeof (mar_cpg_name_t));

	error_put:
	hdb_handle_put (&cpg_iteration_handle_t_db, req_lib_cpg_iterationnext->iteration_handle);
	error_exit:
	res_lib_cpg_iterationnext.header.size = sizeof (res_lib_cpg_iterationnext);
	res_lib_cpg_iterationnext.header.id = MESSAGE_RES_CPG_ITERATIONNEXT;
	res_lib_cpg_iterationnext.header.error = error;

	api->ipc_response_send (conn, &res_lib_cpg_iterationnext,
	sizeof (res_lib_cpg_iterationnext));
	}

	static void message_handler_req_lib_cpg_iteration_finalize (
	void *conn,
	const void *message)
	{
	const struct req_lib_cpg_iterationfinalize *req_lib_cpg_iterationfinalize = message;
	struct res_lib_cpg_iterationfinalize res_lib_cpg_iterationfinalize;
	struct cpg_iteration_instance *cpg_iteration_instance;
	cs_error_t error = CS_OK;
	int res;

	log_printf (LOGSYS_LEVEL_DEBUG, "cpg iteration finalize\n");

	res = hdb_handle_get (&cpg_iteration_handle_t_db,
	req_lib_cpg_iterationfinalize->iteration_handle,
	(void *)&cpg_iteration_instance);

	if (res != 0) {
	error = CS_ERR_LIBRARY;
	goto error_exit;
	}

	assert (cpg_iteration_instance);

	cpg_iteration_instance_finalize (cpg_iteration_instance);
	hdb_handle_put (&cpg_iteration_handle_t_db, cpg_iteration_instance->handle);

	error_exit:
	res_lib_cpg_iterationfinalize.header.size = sizeof (res_lib_cpg_iterationfinalize);
	res_lib_cpg_iterationfinalize.header.id = MESSAGE_RES_CPG_ITERATIONFINALIZE;
	res_lib_cpg_iterationfinalize.header.error = error;

	api->ipc_response_send (conn, &res_lib_cpg_iterationfinalize,
	sizeof (res_lib_cpg_iterationfinalize));
	}

File Metadata

Mime Type: text/x-diff
Expires: Wed, Feb 26, 11:55 PM (11 h, 2 m ago)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1455866
Default Alt Text: (186 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions