/* 
 * ccm.c: Consensus Cluster Service Program 
 *
 * Copyright (c) International Business Machines  Corp., 2002
 * Author: Ram Pai (linuxram@us.ibm.com)
 * 
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 */
#include <lha_internal.h>

#include <ccm.h>
#ifdef HAVE_STDINT_H
#include <stdint.h>
#endif
#include <clplumbing/cl_signal.h>
#include <clplumbing/coredumps.h>
#include "ccmmsg.h"
#include "ccmmisc.h"


/* PROTOTYPE */
static void
ccm_reset_all_join_request(ccm_info_t* info);
static void report_reset(void);
static int ccm_already_joined(ccm_info_t *);
static void ccm_memcomp_reset(ccm_info_t *);

/* For enhanced membership service */
static void append_change_msg(ccm_info_t *info,const char *node);
static int received_all_change_msg(ccm_info_t *info);
static int is_expected_change_msg(ccm_info_t *info, const char *node,
		enum change_event_type);
static void add_change_msg(ccm_info_t *info, const char *node, 
		const char *orig, enum change_event_type);
static void reset_change_info(ccm_info_t *info); 
static void send_mem_list_to_all(ll_cluster_t *hb, ccm_info_t *info, 
		char *cookie);
static void ccm_fill_update_table(ccm_info_t *info, 
		ccm_update_t *update_table, const void *uptime_list);

static void dump_mbrs(ccm_info_t *info);

static longclock_t change_time;
static gboolean gl_membership_converged = FALSE;

const char state_strings[12][64]={
	"CCM_STATE_NONE",
	"CCM_STATE_VERSION_REQUEST",
	"CCM_STATE_JOINING",  		
	"CCM_STATE_SENT_MEMLISTREQ",					
	"CCM_STATE_MEMLIST_RES",				
	"CCM_STATE_JOINED",    
	"CCM_STATE_WAIT_FOR_MEM_LIST",
	"CCM_STATE_WAIT_FOR_CHANGE",
	"CCM_STATE_NEW_NODE_WAIT_FOR_MEM_LIST",
	"CCM_STATE_END"		
};

const char*
state2string(int state){
	if (state > CCM_STATE_END){
		return "INVALID STATE";
	}
	
	return state_strings[state];
}

static int
string2state(const char* state_str)
{
	int  i; 
	
	if (state_str == NULL){
		ccm_log(LOG_ERR, "%s: state_str is NULL", __FUNCTION__);
		return -1;
	}

	for (i = 0 ; i < DIMOF(state_strings); i++){
		if (strncmp(state_strings[i], state_str, 64) == 0){
			return i;
		}		
	}
	
	ccm_log(LOG_ERR, "%s: Cannot find a match for string %s", 
	       __FUNCTION__, state_str);

	return -1;
	
}

static void
ccm_set_state(ccm_info_t* info, int istate,const struct ha_msg*  msg)	
{	
	int oldstate = info->state;								
	info->state = (istate);		
	if((istate)==CCM_STATE_JOINING){
		client_influx();	
	}		
	if (istate == CCM_STATE_JOINED){
		gl_membership_converged =TRUE;
	}
	if (llm_get_myindex(CCM_GET_LLM(info)) == info->ccm_cluster_leader
		   && CCM_STATE_JOINED == istate) {
		info->has_quorum = ccm_calculate_quorum(info);
	}
	else {
		ccm_stop_query_quorum ();
	}
	ccm_debug(LOG_DEBUG,"node state %s -> %s"
	,	state2string(oldstate),state2string(istate)); 
}




static void
change_time_init(void)
{
	change_time = ccm_get_time();
}
static int
change_timeout(unsigned long timeout)
{
	return(ccm_timeout(change_time, ccm_get_time(), timeout));
}

static longclock_t mem_list_time;
static void
mem_list_time_init(void)
{
	mem_list_time = ccm_get_time();
}
static int
mem_list_timeout(unsigned long timeout)
{
	return(ccm_timeout(mem_list_time, ccm_get_time(), timeout));
}

static longclock_t  new_node_mem_list_time;
static void new_node_mem_list_time_init(void)
{
    new_node_mem_list_time = ccm_get_time();
}
static int new_node_mem_list_timeout(unsigned long timeout)
{
    return(ccm_timeout(new_node_mem_list_time, ccm_get_time(), timeout));
}

#define CCM_GET_MYNODE_ID(info) \
	info->llm.nodes[info->llm.myindex].nodename
#define CCM_GET_CL_NODEID(info) \
	info->llm.nodes[CCM_GET_CL(info)].nodename 
#define CCM_GET_RECEIVED_CHANGE_MSG(info, node)				\
	llm_get_change(CCM_GET_LLM(info),llm_get_index(&info->llm, node))
#define CCM_SET_RECEIVED_CHANGE_MSG(info, node, value)			\
	llm_set_change(CCM_GET_LLM(info), llm_get_index(&info->llm, node), value)

/*
////////////////////////////////////////////////////////////////
// BEGIN OF Functions associated with CCM token types that are
// communicated accross nodes and their values.
////////////////////////////////////////////////////////////////
*/

static void ccm_state_wait_for_mem_list(enum ccm_type ccm_msg_type, 
			struct ha_msg *reply, 
			ll_cluster_t *hb, 
			ccm_info_t *info);
static void ccm_state_new_node_wait_for_mem_list(enum ccm_type ccm_msg_type, 
	              struct ha_msg *reply, 
	              ll_cluster_t *hb, 
			ccm_info_t *info);





/* END OF TYPE_STR datastructure and associated functions */


/* */
/* timeout configuration function */
/* */
static void
ccm_configure_timeout(ll_cluster_t *hb, ccm_info_t *info)
{
	long keepalive = hb->llc_ops->get_keepalive(hb);

	ccm_debug2(LOG_DEBUG, "%s: keepalive=%ld", __FUNCTION__, keepalive);

	CCM_TMOUT_SET_U(info, 5*keepalive);
	CCM_TMOUT_SET_LU(info, 30*keepalive);
	CCM_TMOUT_SET_VRS(info, 9*keepalive);
	CCM_TMOUT_SET_ITF(info, 18*keepalive);
	CCM_TMOUT_SET_IFF(info, 12*keepalive);
	CCM_TMOUT_SET_FL(info, CCM_TMOUT_GET_ITF(info)+5);
}


/* */
/* timeout_msg_create:  */
/*	fake up a timeout message, which is in the */
/* 	same format as the other messages that are */
/*	communicated across the nodes. */
/* */




#ifdef TIMEOUT_MSG_FUNCTIONS_NEEDED
/* */
/* timeout_msg_done:  */
/*   done with the processing of this message. */
static void
timeout_msg_done(void)
{
	/* nothing to do. */
	return;
}


/* */
/* timeout_msg_del:  */
/*   delete the given timeout message. */
/*   nobody calls this function.  */
/*   someday somebody will call it :) */
static void
timeout_msg_del(void)
{
	ha_msg_del(timeout_msg);
	timeout_msg = NULL;
}
#endif


/* */
/* These are the function that keep track of number of time a version */
/* response message has been dropped. These function are consulted by */
/* the CCM algorithm to determine if a version response message has */
/* to be dropped or not. */
/* */
static int respdrop=0;
#define MAXDROP 3

static int
resp_can_i_drop(void)
{
	if (respdrop >= MAXDROP){
		return FALSE;
	}
	return TRUE;
}

static void
resp_dropped(void)
{
	respdrop++;
}

static void
resp_reset(void)
{
	respdrop=0;
}
/* */
/* End of response processing messages. */
/* */


/* */
/* BEGIN OF functions that track the time since a connectivity reply has */
/* been sent to the leader. */
/* */
static longclock_t finallist_time;

static void
finallist_init(void)
{
	finallist_time = ccm_get_time();
}

static void
finallist_reset(void)
{
	finallist_time = 0;
}

static int
finallist_timeout(unsigned long timeout)
{
	return(ccm_timeout(finallist_time, ccm_get_time(), timeout));
}
/* */
/* END OF functions that track the time since a connectivity reply has */
/* been sent to the leader. */
/* */


/* Reset all the datastructures. Go to a state which is equivalent */
/* to a state when the node is just about to join a cluster. */
void 
ccm_reset(ccm_info_t *info)
{

	if(ccm_already_joined(info)){
		client_evicted();
	}

	ccm_mem_reset(info);
	ccm_memcomp_reset(info);
	CCM_SET_ACTIVEPROTO(info, CCM_VER_NONE);
	CCM_SET_COOKIE(info,"");
	CCM_SET_MAJORTRANS(info,0);
	CCM_SET_MINORTRANS(info,0);
	CCM_SET_CL(info,-1);
	CCM_SET_JOINED_TRANSITION(info, 0);
	ccm_set_state(info, CCM_STATE_NONE, NULL);
	info->has_quorum = -1;
	update_reset(CCM_GET_UPDATETABLE(info));
	ccm_reset_all_join_request(info);
	version_reset(CCM_GET_VERSION(info));
	finallist_reset();
	leave_reset();
	report_reset();
}

static void 
ccm_init(ccm_info_t *info)
{
	update_init(CCM_GET_UPDATETABLE(info));
	ccm_reset_all_join_request(info);
	CCM_INIT_MAXTRANS(info);
        leave_init();
        (void)timeout_msg_init(info);
	ccm_reset(info);
}



/*
 * BEGIN OF ROUTINES THAT REPORT THE MEMBERSHIP TO CLIENTS.
 */
static void
report_reset(void)
{
	return;
}

/* */
/* print and report the cluster membership to clients. */
/* */
static void
report_mbrs(ccm_info_t *info)
{
	int i;
	const char *nodename;

	static struct born_s  {
		int index;
		int bornon;
	}  bornon[MAXNODE];/*avoid making it a 
				stack variable*/
	

	if(ccm_get_memcount(info)==1){
		bornon[0].index  = CCM_GET_MEMINDEX(info,0);
		bornon[0].bornon = CCM_GET_MAJORTRANS(info);
	} else for(i=0; i < ccm_get_memcount(info); i++){
		bornon[i].index = CCM_GET_MEMINDEX(info,i);
		bornon[i].bornon = update_get_uptime(CCM_GET_UPDATETABLE(info), 
						     CCM_GET_LLM(info),
						     CCM_GET_MEMINDEX(info,i));
		if(bornon[i].bornon==0) 
			bornon[i].bornon=CCM_GET_MAJORTRANS(info);
		assert(bornon[i].bornon!=-1);
	}
	
	ccm_debug2(LOG_DEBUG,"\t\t the following are the members " 
	       "of the group of transition=%d",
	       CCM_GET_MAJORTRANS(info));
	
	for (i=0 ;  i < ccm_get_memcount(info); i++) {
		nodename = llm_get_nodename(CCM_GET_LLM(info), 
					  CCM_GET_MEMINDEX(info,i));
		ccm_debug2(LOG_DEBUG,"\t\tnodename=%s bornon=%d", nodename, 
		       bornon[i].bornon);
	}

	/* 
	 * report to clients, the new membership 
	 */
	dump_mbrs(info);
	client_new_mbrship(info,
			   bornon);
	return;
}



/* */
/* generate a random cookie. */
/* NOTE: cookie  is  a  mechanism  of  seperating out the contexts */
/* of  messages  of  partially  partitioned  clusters. */
/* For example, consider  a  case  where   node  A  is  physically */
/* in  the  partition  X  and  partition  Y,  and  but  has joined  */
/* membership  in partition X. It will end up getting ccm protocol */
/* messages  sent  by  members in both the partitions. In order to  */
/* seperate  out  messages  belonging  to  individual partition, a  */
/* random  string  is  used  as  a identifier by each partition to  */
/* identify  its  messages.  In  the above case A will get message  */
/* from  both  the  partitions  but  only listens to messages from  */
/* partition X and drops messages from partition Y. */
/* */
static char *
ccm_generate_random_cookie(void)
{
	char *cookie;
	int i;
	struct timeval tmp;

	cookie = g_malloc(COOKIESIZE*sizeof(char));
	/* g_malloc never returns NULL: assert(cookie); */

	/* seed the random with a random value */
	gettimeofday(&tmp, NULL);
	srandom((unsigned int)tmp.tv_usec); 

	for ( i = 0 ; i < COOKIESIZE-1; i++ ) {
		cookie[i] = random()%(127-'!')+'!';
	}
	cookie[i] = '\0';
	return cookie;
}


static void
ccm_free_random_cookie(char *cookie)
{
	assert(cookie && *cookie);
	g_free(cookie);
}



/* BEGIN OF FUNCTIONS that keep track of connectivity  information  */
/* conveyed by individual members of the cluster. These  functions  */
/* are used by only the cluster leader. Ultimately these connectivity */
/* information is used by the cluster to extract out the members */
/* of the cluster that have total connectivity. */
static int
ccm_memcomp_cmpr(gconstpointer a, gconstpointer b)
{
	return(*((const uint32_t *)a)-*((const uint32_t *)b));
}
static void
ccm_memcomp_free(gpointer data, gpointer userdata)
{
	if(data) {
		g_free(data);
	}
	return;
}

static void
ccm_memcomp_note(ccm_info_t *info, const char *orig, 
		uint32_t maxtrans, const char *memlist)
{
	int index;
	char *bitmap = NULL;
	uint32_t *ptr;
	memcomp_t *mem_comp = CCM_GET_MEMCOMP(info);

	bitmap_create(&bitmap, MAXNODE);
	if (bitmap == NULL){
		ccm_log(LOG_ERR, "bitmap creatation failed");
		return;
	}
	/* find the index of the originator */
	index = llm_get_index(CCM_GET_LLM(info), orig);
	
	/* convert the memlist into a bit map and feed it to the graph */
	ccm_str2bitmap(memlist, strlen(memlist), bitmap);
	
	graph_update_membership(MEMCOMP_GET_GRAPH(mem_comp), 
				index, bitmap);
	/*NOTE DO NOT DELETE bitlist, because it is 
	 * being handled by graph*/

	ptr = (uint32_t *)g_malloc(2*sizeof(uint32_t));
	ptr[0] = maxtrans;
	ptr[1] = index;
	MEMCOMP_SET_MAXT(mem_comp, 
		(g_slist_insert_sorted(MEMCOMP_GET_MAXT(mem_comp), 
			ptr, ccm_memcomp_cmpr)));
	return;
}

/* called by the cluster leader only  */
static void
ccm_memcomp_note_my_membership(ccm_info_t *info)
{
	char memlist[MAX_MEMLIST_STRING];

	update_strcreate(CCM_GET_UPDATETABLE(info),
			memlist, CCM_GET_LLM(info));
	ccm_memcomp_note(info, llm_get_mynodename(&info->llm), 
			CCM_GET_MAXTRANS(info), memlist);
	return;
}

/* add a new member to the membership list */
static void
ccm_memcomp_add(ccm_info_t *info, const char *orig)
{
	int index, myindex;
	memcomp_t *mem_comp = CCM_GET_MEMCOMP(info);

	index = llm_get_index(CCM_GET_LLM(info), orig);
	myindex = llm_get_myindex(&info->llm); 
	graph_add_uuid(MEMCOMP_GET_GRAPH(mem_comp), index);
	graph_add_to_membership(MEMCOMP_GET_GRAPH(mem_comp), 
				myindex, index);
	/* ccm_memcomp_note(info, orig, maxtrans, memlist); */
	return;
}

static void 
ccm_memcomp_init(ccm_info_t *info)
{
	int track=-1;
	int index;
	
	memcomp_t *mem_comp = CCM_GET_MEMCOMP(info);

	MEMCOMP_SET_GRAPH(mem_comp, graph_init());

	/* go through the update list and note down all the members who
	 * had participated in the join messages. We should be expecting
	 * reply memlist bitmaps atleast from these nodes.
	 */
	while((index = update_get_next_index(CCM_GET_UPDATETABLE(info), 
				CCM_GET_LLM(info), &track)) != -1) {
		graph_add_uuid(MEMCOMP_GET_GRAPH(mem_comp),index); 
	}
	MEMCOMP_SET_MAXT(mem_comp,  NULL);
	MEMCOMP_SET_INITTIME(mem_comp, ccm_get_time());
}


static void 
ccm_memcomp_reset(ccm_info_t *info)
{
	memcomp_t *mem_comp = CCM_GET_MEMCOMP(info);

	graph_free(MEMCOMP_GET_GRAPH(mem_comp));
	MEMCOMP_SET_GRAPH(mem_comp,NULL);
	g_slist_foreach(MEMCOMP_GET_MAXT(mem_comp), 
			ccm_memcomp_free, NULL);
	g_slist_free(MEMCOMP_GET_MAXT(mem_comp));
	MEMCOMP_SET_MAXT(mem_comp,  NULL);
	return;
}


static int
ccm_memcomp_rcvd_all(ccm_info_t *info)
{
	return graph_filled_all(MEMCOMP_GET_GRAPH(CCM_GET_MEMCOMP(info)));
}

static int
ccm_memcomp_timeout(ccm_info_t *info, long timeout)
{
	memcomp_t *mem_comp = CCM_GET_MEMCOMP(info);

	return(ccm_timeout(MEMCOMP_GET_INITTIME(mem_comp), 
				ccm_get_time(), timeout));
}

static int
ccm_memcomp_get_maxmembership(ccm_info_t *info, char **bitmap)
{
	GSList *head;
	uint32_t *ptr;
	int 	uuid;

	memcomp_t *mem_comp = CCM_GET_MEMCOMP(info);

	(void)graph_get_maxclique(MEMCOMP_GET_GRAPH(mem_comp), 
			bitmap);

	head = MEMCOMP_GET_MAXT(mem_comp);

	while (head) {
		ptr = (uint32_t *)g_slist_nth_data(head, 0);
		uuid = ptr[1];
		if(bitmap_test(uuid, *bitmap, MAXNODE)) {
			return ptr[0];
		}
		head = g_slist_next(head);
	}
	return 0;
}


/* */
/* END OF the membership tracking functions. */
/* */



static int
ccm_am_i_leader(ccm_info_t *info)
{
	llm_info_t *llm = CCM_GET_LLM(info);
	
	if ( llm_get_myindex(llm) == CCM_GET_CL(info)){
		return TRUE;
	}
	
	return FALSE;
}

static gboolean
node_is_leader(ccm_info_t* info, const char* nodename)
{
	return( llm_get_index(&info->llm, nodename) == CCM_GET_CL(info));	
	
}

static int
ccm_already_joined(ccm_info_t *info)
{
	if (CCM_GET_JOINED_TRANSITION(info)) {
		return TRUE;
	}
	return FALSE;
}

/* 
 * END  OF  FUNCTIONS  that  keep track of stablized membership list 
 */


/* 
 * BEGIN OF FUNCTIONS THAT KEEP TRACK of cluster nodes that have shown 
 * interest in joining the cluster. 
 *
 * NOTE: when a new node wants to join the cluster, it multicasts a  
 * message asking for the necessary information to send out a  join 
 * message. (it needs the current major transistion number, the context 
 * string i.e cookie, the protocol number that everybody is operating 
 * in). 
 * 
 * The functions below track these messages sent out by new potential 
 * members showing interest in acquiring the initial context. 
 */
static void 
ccm_add_new_joiner(ccm_info_t *info, const char *orig, struct ha_msg* msg)
{
	llm_info_t* llm = &info->llm;
	
	int idx = llm_get_index(&info->llm, orig);
	
	const char* major_trans = 0;
	int trans_val;
	
	/* get the major transition version */
	if ((major_trans = ha_msg_value(msg, CCM_MAJORTRANS)) == NULL) { 
	  ccm_debug(LOG_WARNING, "ccm_state_version_request: "
		    "no protocol information");
	  return;
	}
	
	trans_val = atoi(major_trans);
	
	llm_set_joinrequest(llm, idx, TRUE, trans_val);
	
	return;
}


static gboolean
ccm_get_all_active_join_request(ccm_info_t* info)
{	
	
	llm_info_t* llm = &info->llm;
	size_t i;
	
	for (i = 0 ; i < llm->nodecount; i++){
		if (STRNCMP_CONST(llm->nodes[i].status,"dead") != 0
		    && llm_get_joinrequest(llm, i) == FALSE ){
			return FALSE;
		}
	}
	
	return TRUE;
	
}


static void
ccm_reset_all_join_request(ccm_info_t* info)
{
	llm_info_t* llm = &info->llm;
	size_t i;
	
	for (i = 0 ; i < llm->nodecount; i++){
	  llm_set_joinrequest(llm, i, FALSE, 0);
	}	
}


static int
ccm_am_i_highest_joiner(ccm_info_t *info)
{

	llm_info_t*	llm = &info->llm;
	int		total_nodes =llm->nodecount;
	int		my_indx = llm->myindex;
	int		i;
	
	for (i =0; i < total_nodes;i++){
	  if (i == my_indx) continue;
	  if ( llm_get_joinrequest(llm, i)){
	    int major_trans =llm_get_joinrequest_majortrans(llm, i);
	    int my_major_trans = CCM_GET_MAJORTRANS(info);
	    if (major_trans > my_major_trans ){
	      return FALSE;
	    }else if (major_trans == my_major_trans){
	      if (i > my_indx){
		return FALSE;
	      }
	    }
	  }

	}

	return TRUE;
}

static void 
ccm_remove_new_joiner(ccm_info_t *info, const char *orig)
{
	llm_info_t* llm = &info->llm;
	int index = llm_get_index(llm, orig);
	
	llm_set_joinrequest(llm, index, FALSE, 0);
	
	return;
}




/* send reply to a join quest and clear the request*/

static void 
ccm_send_join_reply(ll_cluster_t *hb, ccm_info_t *info)
{
	llm_info_t* llm = &info->llm;
	size_t i;
	
	for (i = 0 ; i < llm->nodecount; i++){
		if ( i == (size_t)llm->myindex){
			continue;
		}
		if (llm_get_joinrequest(llm, i)){
			ccm_send_one_join_reply(hb,info, llm->nodes[i].nodename);
			llm_set_joinrequest(llm, i, FALSE, 0);
		}
	}
}


/* */
/* END OF FUNCTIONS THAT KEEP TRACK of cluster nodes that have shown */
/* interest in joining the cluster. */
/* */


/*
/////////////////////////////////////////////////////////////////////
//
// BEGIN OF FUNCTIONS THAT SEND OUT messages to nodes of the cluster
//
/////////////////////////////////////////////////////////////////////
*/

/* compute the final membership list from the acquired connectivity */
/* information from other nodes. And send out the consolidated */
/* members of the cluster information to the all the members of  */
/* that have participated in the CCM protocol. */
/* */
/* NOTE: Called by the cluster leader only. */
/* */
static void
ccm_compute_and_send_final_memlist(ll_cluster_t *hb, ccm_info_t *info)
{
	char *bitmap;
	uint maxtrans;
	char string[MAX_MEMLIST_STRING];
	char *cookie = NULL;
	int repeat;

	/* get the maxmimum membership list */
	maxtrans = ccm_memcomp_get_maxmembership(info, &bitmap);

	
	/* create a string with the membership information */
	ccm_bitmap2str(bitmap,  string, MAX_MEMLIST_STRING);
	

	cookie = ccm_generate_random_cookie();

	repeat = 0;
	ccm_mem_bitmapfill(info, bitmap);
	bitmap_delete(bitmap);
	
	while (ccm_send_final_memlist(hb, info, cookie, string, maxtrans+1) 
					!= HA_OK) {
		if(repeat < REPEAT_TIMES){
			ccm_log(LOG_ERR,
				"%s: failure to send finalmemlist", __FUNCTION__);
			cl_shortsleep();
			repeat++;
		}else{
			bitmap_delete(bitmap);
			return;
		}
	}

	/* fill my new memlist and update the new cookie if any */

	/* increment the major transition number and reset the
	 * minor transition number
	 */
	CCM_SET_MAJORTRANS(info, maxtrans+1); 
	CCM_RESET_MINORTRANS(info);

	/* if cookie has changed update it.
	 */
	if (cookie) {
		ccm_debug2(LOG_DEBUG, "%s: cookie changed ", __FUNCTION__);
		CCM_SET_COOKIE(info, cookie); 
		ccm_free_random_cookie(cookie);
	}

	/* check if any joiner is waiting for a response from us. 
	 * If so respond and free all the joiners.
	 */
	ccm_send_join_reply(hb, info);

	CCM_SET_CL(info, llm_get_myindex(CCM_GET_LLM(info)));
	report_mbrs(info);/* call this before update_reset() */
/*	update_reset(CCM_GET_UPDATETABLE(info));*/
	ccm_memcomp_reset(info);
	ccm_set_state(info, CCM_STATE_JOINED, NULL);
	if(!ccm_already_joined(info)) {
		CCM_SET_JOINED_TRANSITION(info, CCM_GET_MAJORTRANS(info));
	}

	return;
}







/* */
/* Browse through the list of all the connectivity request messages */
/* from cluster leaders. Send out the connectivity information only */
/* to the node which we believe is the cluster leader. To everybody  */
/* else send out a null message. */
/* */
static int
ccm_send_cl_reply(ll_cluster_t *hb, ccm_info_t *info)
{
	int ret=FALSE;
	char memlist[MAX_MEMLIST_STRING];
	const char* cl;
	const char* cl_tmp;
	void *cltrack;
	uint  trans;
	int repeat;
	/*
        * Get the name of the cluster leader
	*/
	cl = update_get_cl_name(CCM_GET_UPDATETABLE(info), 
				CCM_GET_LLM(info));

	/* search through the update list and find if any Cluster
	 * leader has sent a memlist request. For each, check if
	 * that node is the one which we believe is the leader.
	 * if it is the leader, send it our membership list.
	 * if not send it an NULL membership reply.
	 */
	cltrack = update_initlink(CCM_GET_UPDATETABLE(info));
	while((cl_tmp = update_next_link(CCM_GET_UPDATETABLE(info), 
				CCM_GET_LLM(info), cltrack, &trans)) != NULL) {
		if(strncmp(cl, cl_tmp, 
			   NODEIDSIZE) == 0) {

			if(ccm_already_joined(info) && 
				CCM_GET_MAJORTRANS(info) != trans){
				ccm_log(LOG_INFO, "ccm evicted");
				ccm_reset(info);
				return FALSE;
			}
			ret = TRUE;
			update_strcreate(CCM_GET_UPDATETABLE(info), 
							 memlist, CCM_GET_LLM(info));

			/* send Cluster Leader our memlist only if we are 
			 * operating in the same transition as that of 
			 * the leader, provided we have been a cluster member 
			 * in the past 
			 */
			repeat = 0;
			while (ccm_send_memlist_res(hb, info, cl, memlist)
						!=HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_log(LOG_ERR,
						"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}
		} else {
			/* I dont trust this Cluster Leader.
			Send NULL memlist message */
			repeat = 0;
			while (ccm_send_memlist_res(hb, info, cl_tmp, NULL)
					!= HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_log(LOG_ERR, 
					"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}
		}
	}
	update_freelink(CCM_GET_UPDATETABLE(info), cltrack);
	update_free_memlist_request(CCM_GET_UPDATETABLE(info)); 
	return ret;
}
/*
/////////////////////////////////////////////////////////////////////
//
// END OF FUNCTIONS THAT SEND OUT messages to nodes of the cluster
//
/////////////////////////////////////////////////////////////////////
*/


struct ha_msg * ccm_readmsg(ccm_info_t *info, ll_cluster_t *hb);


struct ha_msg *
ccm_readmsg(ccm_info_t *info, ll_cluster_t *hb)
{
	int 	uuid;

	assert(hb);

	/* check if there are any leave events to be delivered */
	if ((uuid=leave_get_next()) != -1) {
		/* create a leave message and return it */
		return ccm_create_leave_msg(info, uuid);
	}
	
	return hb->llc_ops->readmsg(hb, 0);
}



/* */
/* Move the state of this ccm node, from joining state directly to */
/* the joined state. */
/* */
/* NOTE: this is generally called when a joining nodes determines */
/* that it is the only node in the cluster, and everybody else are */
/* dead. */
/* */
static void
ccm_joining_to_joined(ll_cluster_t *hb, ccm_info_t *info)
{
	char *bitmap;
	char *cookie = NULL;

	/* create a bitmap with the membership information */
	(void) bitmap_create(&bitmap, MAXNODE);
	bitmap_mark(llm_get_myindex(&info->llm), bitmap, MAXNODE);

	/* 
	 * I am the only around! Lets discard any cookie that we
	 * got from others, and create a new cookie.
	 * This bug was noticed: when testing with partitioned
	 * clusters.
	 */
	cookie = ccm_generate_random_cookie();

	/* fill my new memlist and update the new cookie if any */
	ccm_mem_bitmapfill(info, bitmap);
	bitmap_delete(bitmap);

	/* increment the major transition number and reset the
	 * minor transition number
	 */
	CCM_INCREMENT_MAJORTRANS(info); 
	CCM_RESET_MINORTRANS(info);

	/* if cookie has changed update it.
	 */
	if (cookie) {
		ccm_debug2(LOG_DEBUG, "%s: cookie changed ", __FUNCTION__);
		CCM_SET_COOKIE(info, cookie); 
		ccm_free_random_cookie(cookie);
	}

	/* check if any joiner is waiting for a response from us. 
	 * If so respond 
	 */
	ccm_send_join_reply(hb, info);
	
	CCM_SET_CL(info, llm_get_myindex(CCM_GET_LLM(info)));
	update_reset(CCM_GET_UPDATETABLE(info));
	ccm_set_state(info, CCM_STATE_JOINED, NULL);
	report_mbrs(info);
	if(!ccm_already_joined(info)) {
		CCM_SET_JOINED_TRANSITION(info, 1);
	}
	return;
}

/*
 * Move the state of this ccm node, from init state directly to 
 * the joined state. 
 * 
 * NOTE: this is generally called when a node when it  determines 
 * that it is all alone in the cluster. 
 */

static int
ccm_init_to_joined(ccm_info_t *info)
{
	char*		cookie;
	int		ret;
	llm_info_t*	llm = &info->llm;
	
	ccm_mem_reset(info);
	ret = ccm_mem_add(info, llm_get_myindex(llm));
	if (ret != HA_OK){
		ccm_log(LOG_ERR, "%s: adding myself to membership failed",
		       __FUNCTION__);
		return HA_FAIL;
	}
	
	llm_set_uptime(llm, llm_get_myindex(llm), 1);

	CCM_SET_MAJORTRANS(info, CCM_GET_MAJORTRANS(info)+1);
	CCM_SET_MINORTRANS(info, 0);
	cookie = ccm_generate_random_cookie();
	CCM_SET_COOKIE(info, cookie);
	ccm_free_random_cookie(cookie);
	CCM_SET_CL(info, llm_get_myindex(CCM_GET_LLM(info)));
	ccm_set_state(info, CCM_STATE_JOINED, NULL);
	CCM_SET_JOINED_TRANSITION(info, 1);
	report_mbrs(info);
	return HA_OK;
}





static void
ccm_all_restart(ll_cluster_t* hb, ccm_info_t* info, struct ha_msg* msg)
{
	const char * orig;
	llm_info_t* llm = & info->llm;

	if ( (orig = ha_msg_value(msg, F_ORIG)) ==NULL){
		ccm_log(LOG_ERR, "orig not found in message");
		return ; 
	}
	
	if (strncmp(orig, llm_get_mynodename(llm), NODEIDSIZE) == 0){
		/*don't react to our own message*/
		return ;
	}
	
	if (info->state != CCM_STATE_VERSION_REQUEST
	    && gl_membership_converged ){
		gl_membership_converged = FALSE;
		if (ccm_get_memcount(info) == 1) {
			ccm_reset(info);
			return;
		}
		ccm_set_state(info, CCM_STATE_NONE, msg);
		ccm_reset_all_join_request(info);
		CCM_SET_CL(info,-1);
		if (ccm_send_restart_msg(hb, info) != HA_OK){
			ccm_log(LOG_ERR, "sending out restart msg failed");
			return;
		}
		
		if (ccm_send_protoversion(hb, info) != HA_OK){
			ccm_log(LOG_ERR, "sending protoversion failed");
			return;
		}
		ccm_set_state(info, CCM_STATE_VERSION_REQUEST, NULL);
	}
	
}

static int
ccm_handle_state_info(ll_cluster_t* hb, ccm_info_t* info, struct ha_msg* msg)
{
	const char* other_node_state;
	int state;
	
	if (!part_of_cluster(info->state)){
		return HA_OK;
	}
	
	other_node_state = ha_msg_value(msg, F_STATE);
	state =  string2state(other_node_state);
	
	if (state < 0){
		ccm_log(LOG_ERR, "%s: wrong state", __FUNCTION__);
		return HA_FAIL;
	}
	
	if (!part_of_cluster(state)){
		return HA_OK;
	}
	
	/*both machines are already part of a cluster, 
	  i.e. we are merging two partitions
	*/
	ccm_all_restart(hb, info, msg);
	
	return HA_OK;
		
}

/* */
/* The state machine that processes message when it is */
/*	the CCM_STATE_VERSION_REQUEST state */
/* */
static void
ccm_state_version_request(enum ccm_type ccm_msg_type,
			struct ha_msg *reply,
			ll_cluster_t *hb, 
			ccm_info_t *info)
{
	const char *orig, *proto, *cookie, *trans, *clsize;
	uint trans_val;
	int  proto_val;
	uint clsize_val;
	int try;
	int repeat;
	
	/* who sent this message */
	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown", __FUNCTION__);
		return;
	}

	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) { 
		ccm_debug(LOG_WARNING, "%s: received message from unknown host %s",
		__FUNCTION__, orig);
		return;
	}

	switch (ccm_msg_type)  {

	case CCM_TYPE_PROTOVERSION_RESP:

		/* get the protocol version */
		if ((proto = ha_msg_value(reply, CCM_PROTOCOL)) == NULL) {
			ccm_debug(LOG_WARNING, "%s: no protocol information", __FUNCTION__);
			return;
		}

		proto_val = atoi(proto); /*TOBEDONE*/
		if (proto_val >= CCM_VER_LAST) {
			ccm_debug(LOG_WARNING, "%s: unknown protocol value", __FUNCTION__);
			ccm_reset(info);
			return;
		}


		/* if this reply has come from a node which is a member
		 * of a larger cluster, we will try to join that cluster
		 * else we will wait for some time, by dropping this
		 * response.
		 */
		if(resp_can_i_drop()) {
			if ((clsize = ha_msg_value(reply, CCM_CLSIZE)) == NULL){
				ccm_debug(LOG_WARNING, "%s: no cluster size information", __FUNCTION__);
				return;
			}
			clsize_val = atoi(clsize);
			if((clsize_val+1) <=  
			   (llm_get_nodecount(CCM_GET_LLM(info))+1)/2) {
				/* drop the response. We will wait for 
			  	 * a response from a bigger group 
				 */
				resp_dropped();
				cl_shortsleep(); /* sleep for a while */
				/* send a fresh version request message */
				version_reset(CCM_GET_VERSION(info));
				ccm_set_state(info, CCM_STATE_NONE, reply);
				/* free all the joiners that we accumulated */
				ccm_reset_all_join_request(info);
				break;
			} 
		}
		resp_reset();
	

		/* get the cookie string */
		if ((cookie = ha_msg_value(reply, CCM_COOKIE)) == NULL) {
			ccm_debug(LOG_WARNING, "%s: no cookie information", __FUNCTION__);
			return;
		}

		/* get the major transition version */
		if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) { 
			ccm_debug(LOG_WARNING, "%s: no protocol information", __FUNCTION__);
			return;
		}

		trans_val = atoi(trans);

		/* send the alive message to the cluster
		    The alive msg means: "I want to join this partition!"*/
		CCM_SET_ACTIVEPROTO(info, proto_val);
		CCM_SET_MAJORTRANS(info, trans_val);
		CCM_SET_MINORTRANS(info, 0);
		CCM_SET_COOKIE(info, cookie);
		version_set_nresp(CCM_GET_VERSION(info),0);
		repeat = 0;
		while(ccm_send_alive_msg(hb, info) != HA_OK){
			if(repeat < REPEAT_TIMES){
				ccm_debug(LOG_WARNING, "%s: failure to send alive", __FUNCTION__);
				cl_shortsleep();
				repeat++;
			}else{
				break;
			}
		}

		/* initialize the update table  
			and set our state to NEW_NODE_WAIT_FOR_MEM_LIST */
		update_reset(CCM_GET_UPDATETABLE(info));
		new_node_mem_list_time_init();
		ccm_set_state(info, CCM_STATE_NEW_NODE_WAIT_FOR_MEM_LIST, reply);

		/* free all the joiners that we accumulated */
		ccm_reset_all_join_request(info);

		break;

	case CCM_TYPE_TIMEOUT:
		try = version_retry(CCM_GET_VERSION(info), 
					CCM_TMOUT_GET_VRS(info));
		switch (try) {
		case VER_NO_CHANGE: 
			break;
		case VER_TRY_AGAIN:
			ccm_set_state(info, CCM_STATE_NONE, reply);
			break;
		case VER_TRY_END:
			if(ccm_am_i_highest_joiner(info)) {
				ccm_init_to_joined(info);
				ccm_send_join_reply(hb, info);
				
			} else {
				ccm_debug2(LOG_DEBUG,"joined but not really");
				version_reset(CCM_GET_VERSION(info));
				ccm_set_state(info, CCM_STATE_NONE, reply);
				ccm_reset_all_join_request(info);
			}
			break;
		}
		break;
				
	case CCM_TYPE_PROTOVERSION:
		/*
		 * cache this request. If we declare ourselves as
		 * a single member group, and if we find that
		 * somebody else also wanted to join the group.
		 * we will restart the join.
		 */

		ccm_add_new_joiner(info, orig, reply);
		if (ccm_get_all_active_join_request(info)
		    && ccm_am_i_highest_joiner(info)){
			ccm_init_to_joined(info);
			ccm_send_join_reply(hb, info);
		}
		
		break;


	case CCM_TYPE_ABORT:
		/* note down there is some activity going 
		 * on and we are not yet alone in the cluster 
		 */
		version_some_activity(CCM_GET_VERSION(info));
	default:
		/* nothing to do. Just forget the message */
		break;
	}

	return;
}


static void
ccm_state_none(enum ccm_type msgtype,
	       struct ha_msg *msg,
	       ll_cluster_t *hb, 
	       ccm_info_t *info)
{
	
	if (ccm_send_protoversion(hb, info)!= HA_OK){
		ccm_log(LOG_ERR, "sending version message failed");
		return;
	}
	
	ccm_set_state(info, CCM_STATE_VERSION_REQUEST, NULL);	
	
	ccm_state_version_request(msgtype, msg, hb, info);
}


/* */
/* The state machine that processes message when it is */
/*	CCM_STATE_JOINED state. */
/* */
static void
ccm_state_joined(enum ccm_type ccm_msg_type, 
			struct ha_msg *reply, 
			ll_cluster_t *hb, 
			ccm_info_t *info)
{
	const char *orig,  *trans, *uptime;
	uint  trans_majorval=0, trans_minorval=0, uptime_val;
	int repeat;
	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown"
		,	__FUNCTION__);
		return;
	}
	
	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) {
		ccm_debug(LOG_WARNING, "%s: received message "
				"from unknown host %s", __FUNCTION__, orig);
		return;
	}
	
	if(ccm_msg_type != CCM_TYPE_PROTOVERSION
	   && ccm_msg_type !=  CCM_TYPE_STATE_INFO
	   && ccm_msg_type != CCM_TYPE_RESTART) {
		const char* tmpcookie =  ha_msg_value(reply, CCM_COOKIE);
		if (tmpcookie == NULL){
			abort();
		}
		if(strncmp(CCM_GET_COOKIE(info), 
			ha_msg_value(reply, CCM_COOKIE), COOKIESIZE) != 0){
			ccm_debug(LOG_WARNING, "%s: received message "
			       "with unknown cookie, just dropping", __FUNCTION__);
			dump_mbrs(info);
			return;
		}



		/* get the major transition version */
		if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) { 
			ccm_debug(LOG_WARNING, "%s: no transition major "
				"information", __FUNCTION__);
			return;
		}
		trans_majorval = atoi(trans);

	 	/*drop the message if it has lower major transition number */
		if (CCM_TRANS_EARLIER(trans_majorval,  
					CCM_GET_MAJORTRANS(info))) {
			ccm_debug(LOG_WARNING, "%s: received "
				"%s message with "
				"a earlier major transition number "
				"recv_trans=%d, mytrans=%d", __FUNCTION__,
				ccm_type2string(ccm_msg_type), trans_majorval, 
				CCM_GET_MAJORTRANS(info));
			return;
		}


		/* get the minor transition version */
		if ((trans = ha_msg_value(reply, CCM_MINORTRANS)) == NULL) { 
			ccm_debug(LOG_WARNING, "%s: no transition minor "
					"information", __FUNCTION__);
			return;
		}

		trans_minorval = atoi(trans);
	}

	switch (ccm_msg_type)  {

		case CCM_TYPE_PROTOVERSION_RESP:
			ccm_debug(LOG_WARNING, "%s: dropping message "
				"of type %s.  Is this a Byzantine failure?", 
					__FUNCTION__, ccm_type2string(ccm_msg_type));

			break;

		case CCM_TYPE_PROTOVERSION:
			/* If we were leader in the last successful iteration,
 			 * then we shall respond with the neccessary information
			 */
			if (ccm_am_i_leader(info)){
				repeat = 0;
				while (ccm_send_one_join_reply(hb, info, orig)
						!= HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join reply", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
			}
			break;

		case CCM_TYPE_JOIN:
			/* get the update value */
			if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
				ccm_debug(LOG_WARNING, "%s: no update "
						"information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);

			/* update the minor transition number if it is of 
			 * higher value and send a fresh JOIN message 
			 */
			if (trans_minorval < CCM_GET_MINORTRANS(info)) {
				ccm_log(LOG_WARNING,
				"%s: got a join message from %s from lower "
				"transition, restarting", __FUNCTION__, orig);
				ccm_all_restart(hb, info, reply);
				break;
			}
			update_reset(CCM_GET_UPDATETABLE(info));
			update_add(CCM_GET_UPDATETABLE(info), CCM_GET_LLM(info),
						orig, uptime_val, TRUE);

			CCM_SET_MINORTRANS(info, trans_minorval);
			repeat = 0;
			while (ccm_send_join(hb, info) != HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}

			ccm_set_state(info, CCM_STATE_JOINING, reply);
			break;	

		case CCM_TYPE_LEAVE: 
			
			if (!node_is_member(info, orig)){
				return;
			}
			
			/* If the dead node is the partition leader, go to
			 * JOINING state
			 */
			if (node_is_leader(info, orig)){
				update_reset(CCM_GET_UPDATETABLE(info));
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING,reply);
				return;
			}

			/* If I'm the leader, record this "I received the
			 * LEAVE message" and transit to WAIT_FOR_CHANGE
			 */
			if(ccm_am_i_leader(info)){
				reset_change_info(info);
				update_reset(CCM_GET_UPDATETABLE(info));
				add_change_msg(info, orig, CCM_GET_MYNODE_ID(info), 
						NODE_LEAVE);
				update_add(CCM_GET_UPDATETABLE(info), 
						CCM_GET_LLM(info), CCM_GET_MYNODE_ID(info),
						CCM_GET_JOINED_TRANSITION(info), FALSE);
				if(received_all_change_msg(info)){
					char *newcookie = ccm_generate_random_cookie();

					ccm_mem_update(info, orig, NODE_LEAVE);          
					send_mem_list_to_all(hb, info, newcookie);
					CCM_SET_MAJORTRANS(info, trans_majorval+1); 
					CCM_RESET_MINORTRANS(info);
					CCM_SET_COOKIE(info, newcookie); 
					ccm_free_random_cookie(newcookie);
					report_mbrs(info);
					return;
				}
				change_time_init();
				ccm_bcast_node_leave_notice(hb,info, orig);
				ccm_set_state(info, CCM_STATE_WAIT_FOR_CHANGE, reply);	
				
				
			}

			break;
			
		case CCM_TYPE_NODE_LEAVE_NOTICE:{
			const char* node;
			const char*	leader = orig;
			
			node = ha_msg_value(reply, F_NODE);
			if(node == NULL){
				ccm_log(LOG_ERR, "%s: node not found in the message"
				, __FUNCTION__);
				ccm_message_debug2(LOG_INFO, reply);
				return;
			}
			
			if (!node_is_member(info, node)){
				return;
			}			
			
			if( !ccm_am_i_leader(info)){				
				send_node_leave_to_leader(hb, info, leader);
				mem_list_time_init();
				ccm_set_state(info,CCM_STATE_WAIT_FOR_MEM_LIST, reply);
			}
			break;
		}
		case CCM_TYPE_NODE_LEAVE:
			if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
				ccm_debug(LOG_WARNING, "%s: no update "
						"information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);

			/* If I'm leader, record received LEAVE message by orig 
			 * and transition to WAIT_FOR_CHANGE state
			 */
			if(ccm_am_i_leader(info)){           
				const char *node = ha_msg_value(reply, F_NODE);

				reset_change_info(info);
				update_reset(CCM_GET_UPDATETABLE(info));
				add_change_msg(info,node,orig,NODE_LEAVE);
				update_add(CCM_GET_UPDATETABLE(info), 
						CCM_GET_LLM(info), orig, uptime_val, FALSE);
				change_time_init();
				ccm_set_state(info, CCM_STATE_WAIT_FOR_CHANGE, reply);
			}
			break;

		case CCM_TYPE_ALIVE:
			/* If I'm leader, record I received the ALIVE message and 
			 * transit to WAIT_FOR_CHANGE
			 */
			if (ccm_am_i_leader(info)){
				reset_change_info(info);
				update_reset(CCM_GET_UPDATETABLE(info));
				add_change_msg(info,orig, CCM_GET_MYNODE_ID(info), 
                            		NEW_NODE);
				update_add(CCM_GET_UPDATETABLE(info),
                            		CCM_GET_LLM(info), CCM_GET_MYNODE_ID(info),
					CCM_GET_JOINED_TRANSITION(info), FALSE);

				if(received_all_change_msg(info)){
					char *newcookie = ccm_generate_random_cookie();

					ccm_mem_update(info, orig, NEW_NODE);
					update_add(CCM_GET_UPDATETABLE(info),
						CCM_GET_LLM(info),
						info->change_node_id, trans_majorval+1, FALSE);
					send_mem_list_to_all(hb, info, newcookie);
					CCM_SET_MAJORTRANS(info, trans_majorval+1);
					CCM_RESET_MINORTRANS(info);
					CCM_SET_COOKIE(info, newcookie);
					ccm_free_random_cookie(newcookie);
					report_mbrs(info);
					return;
				}
				change_time_init();
				ccm_set_state(info, CCM_STATE_WAIT_FOR_CHANGE, reply);
			}else{
				/* I'm not leader, send CCM_TYPE_NEW_NODE
				 * to leader and transit to WAIT_FOR_MEM_LIST
				 */
				ccm_send_newnode_to_leader(hb, info, orig);
				mem_list_time_init();
				ccm_set_state(info,CCM_STATE_WAIT_FOR_MEM_LIST, reply);
			}
			break;

		case CCM_TYPE_NEW_NODE:
			if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
				ccm_debug(LOG_WARNING, "%s: no update "
						"information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);

			/* If I'm leader, record received ALIVE message by orig 
			 * and transition to WAIT_FOR_CHANGE state
			 */
			if(ccm_am_i_leader(info)){
				const char *node = ha_msg_value(reply, F_NODE);

				reset_change_info(info);	
				update_reset(CCM_GET_UPDATETABLE(info));
				add_change_msg(info,node, orig, NEW_NODE);
				update_add(CCM_GET_UPDATETABLE(info), 
						CCM_GET_LLM(info),
						orig, uptime_val, FALSE);
				change_time_init();
				ccm_set_state(info, CCM_STATE_WAIT_FOR_CHANGE, reply);
			}
			break;

		case CCM_TYPE_STATE_INFO:
			ccm_handle_state_info(hb, info, reply);
			break;			
		case CCM_TYPE_RESTART:
			ccm_all_restart(hb, info, reply);
			break;
		case CCM_TYPE_MEM_LIST:{
			int quorum;
			const char* memlist;
			if (strncmp(orig, llm_get_mynodename((&info->llm) ), NODEIDSIZE) == 0){
				/*this message is from myself, ignore it*/
				break;
			}

			memlist = ha_msg_value(reply, CCM_MEMLIST);
			if (memlist == NULL){
				break;
			}
			
			if (ha_msg_value_int (reply, CCM_QUORUM, &quorum)==HA_OK){ 
				info->has_quorum = quorum;
			}
			else {
				info->has_quorum = -1;
			}
			
			if (node_is_leader(info, orig)
			    && !am_i_member_in_memlist(info, memlist)){
				ccm_set_state(info, CCM_STATE_NONE, reply);
				break;
			}
			report_mbrs(info);

			break;
		}

		case CCM_TYPE_REQ_MEMLIST:
		case CCM_TYPE_RES_MEMLIST:
		case CCM_TYPE_FINAL_MEMLIST:
		case CCM_TYPE_ABORT:
			ccm_log(LOG_ERR, "%s: dropping message "
				"of type %s. Is this a Byzantine failure?"
			,	__FUNCTION__, ccm_type2string(ccm_msg_type));
			/* nothing to do. Just forget the message */
			break;
		default:
			break;
	}
}

/* */
/* The state machine that processes message when it is in */
/* CCM_STATE_WAIT_FOR_CHANGE state. */
/* */
static void ccm_state_wait_for_change(enum ccm_type ccm_msg_type,
			struct ha_msg *reply,
			ll_cluster_t *hb,
			ccm_info_t *info)
{
	const char *orig, *trans, *uptime, *node;
	uint trans_majorval=0, trans_minorval=0, uptime_val=0;
	gboolean uptime_set = FALSE;
	int repeat;
	
	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown"
		,	__FUNCTION__ );
		return;
	}

	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) {
		ccm_debug(LOG_WARNING, "%s: received message "
				"from unknown host %s", __FUNCTION__, orig);
		return;
	}
	node = ha_msg_value(reply, F_NODE);

	if(ccm_msg_type != CCM_TYPE_PROTOVERSION
	   && ccm_msg_type !=  CCM_TYPE_STATE_INFO
	   && ccm_msg_type != CCM_TYPE_RESTART) {
		
		if(strncmp(CCM_GET_COOKIE(info),
			ha_msg_value(reply, CCM_COOKIE), COOKIESIZE) != 0){
			ccm_debug(LOG_WARNING, "%s: received message "
					"with unknown cookie, just dropping", __FUNCTION__);
			return;
		}

		/* get the major transition version */
		if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) {
			ccm_debug(LOG_WARNING, "%s: no transition major "
				"information", __FUNCTION__);
			return;
		}
		trans_majorval = atoi(trans);

	 	/* drop the message if it has lower major transition number */
		if (CCM_TRANS_EARLIER(trans_majorval,
					CCM_GET_MAJORTRANS(info))) {
			ccm_debug(LOG_WARNING, "%s: received "
				"%s message with "
				"a earlier major transition number "
				"recv_trans=%d, mytrans=%d", __FUNCTION__,
				ccm_type2string(ccm_msg_type), trans_majorval,
				CCM_GET_MAJORTRANS(info));
			return;
		}

		/* get the minor transition version */
		if ((trans = ha_msg_value(reply, CCM_MINORTRANS)) == NULL) { 
			ccm_debug(LOG_WARNING, "%s: no transition minor "
					"information", __FUNCTION__);
			return;
		}
		trans_minorval = atoi(trans);
	}

	switch (ccm_msg_type) {
		case CCM_TYPE_PROTOVERSION:
			/*
			 * cache this request. We will respond to it,
			 * after transition is complete.
			 */
		  ccm_add_new_joiner(info, orig, reply);
			break;
			
		case CCM_TYPE_NODE_LEAVE_NOTICE:
			/* It is my own message, then I can ignore it
			 * or from another lead, i.e. we are in split-brain
			 * and I can do nothing about it
			 */
			break;
		case CCM_TYPE_LEAVE:
			
			if (!node_is_member(info, orig)){
				return;
			}			
			
			if(strcmp(info->change_node_id, orig) == 0
			   && info->change_type == NODE_LEAVE){
				/*It is the same node leaving*/
				return;
			}
			
			node = orig;
			orig = CCM_GET_MYNODE_ID(info);
			uptime_val = CCM_GET_JOINED_TRANSITION(info);
			uptime_set = TRUE;
			/*fall through*/
		case CCM_TYPE_NODE_LEAVE:               
			/* only leader can stay in this state */
			if(!ccm_am_i_leader(info))
				break;

			if (!uptime_set){
				if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
					ccm_debug(LOG_WARNING, "%s: no update information",
					__FUNCTION__);
					return;
				}
				uptime_val = atoi(uptime);
				uptime_set = TRUE;
			}

			/* Record received LEAVE message by orig.
			 * If received all change msg, send mem_list to members.
			 */
			if(is_expected_change_msg(info,node,NODE_LEAVE)){
				append_change_msg(info,orig);
				update_add(CCM_GET_UPDATETABLE(info),
					CCM_GET_LLM(info), orig, uptime_val, FALSE);

				if(received_all_change_msg(info)){ 
					char *newcookie = ccm_generate_random_cookie();

					ccm_mem_update(info, node, NODE_LEAVE);        
     					send_mem_list_to_all(hb, info, newcookie);
					CCM_SET_MAJORTRANS(info, trans_majorval+1); 
					CCM_RESET_MINORTRANS(info);
					CCM_SET_COOKIE(info, newcookie); 
					report_mbrs(info);
					reset_change_info(info); 
/*					update_reset(CCM_GET_UPDATETABLE(info));*/
					ccm_free_random_cookie(newcookie);
					ccm_send_join_reply(hb, info);
					CCM_SET_CL(info, llm_get_myindex(CCM_GET_LLM(info)));
					ccm_set_state(info, CCM_STATE_JOINED,reply);
					return;
				}
			}else{
				reset_change_info(info);
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
				return;
			}                  
			break;
		
		case CCM_TYPE_ALIVE:
			node = orig;	
			orig = CCM_GET_MYNODE_ID(info);
			uptime_val = CCM_GET_JOINED_TRANSITION(info);
			uptime_set = TRUE;
			/*fall through*/	    	
		case CCM_TYPE_NEW_NODE:
			/* only leader can stay in this state */
			if(!ccm_am_i_leader(info)){
				assert(0);
			}

			if (!uptime_set){
				if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
					ccm_debug(LOG_WARNING, "%s: no update information",
					__FUNCTION__);
					return;
				}
				uptime_val = atoi(uptime);
				uptime_set = TRUE;
			}

			if(is_expected_change_msg(info,node, NEW_NODE)){
				append_change_msg(info,orig);
				update_add(CCM_GET_UPDATETABLE(info), CCM_GET_LLM(info),
					orig, uptime_val, FALSE);

				if(received_all_change_msg(info)){
					char *newcookie = ccm_generate_random_cookie();

					ccm_mem_update(info, node, NEW_NODE); 
					update_add(CCM_GET_UPDATETABLE(info), 
						CCM_GET_LLM(info),
						info->change_node_id, trans_majorval+1, FALSE);                                
					send_mem_list_to_all(hb, info, newcookie);
					CCM_SET_MAJORTRANS(info, trans_majorval+1); 
					CCM_RESET_MINORTRANS(info);
					CCM_SET_COOKIE(info, newcookie); 
					report_mbrs(info);
					reset_change_info(info); 
/*					update_reset(CCM_GET_UPDATETABLE(info));*/
					ccm_free_random_cookie(newcookie);
					ccm_send_join_reply(hb, info);
					ccm_set_state(info, CCM_STATE_JOINED, reply);
					return;
				}                       
			}else{
				reset_change_info(info);
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				while (ccm_send_join(hb, info) != HA_OK) {
					ccm_debug(LOG_WARNING, "%s: failure to send join",
					__FUNCTION__);
					cl_shortsleep();
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
				return;
			}
			break;
	    	
		case CCM_TYPE_TIMEOUT:
			if(change_timeout(CCM_TMOUT_GET_U(info))){
				reset_change_info(info);
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
			}
			break;

		case CCM_TYPE_JOIN:
			/* get the update value */
			if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
				ccm_debug(LOG_WARNING, "%s: no update information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);
			uptime_set = TRUE;

			/* update the minor transition number if it is of 
			 * higher value and send a fresh JOIN message 
			 */
			if (trans_minorval < CCM_GET_MINORTRANS(info)) {
				ccm_log(LOG_WARNING,
				"%s: got a join message from %s from lower "
				"transition, restarting", __FUNCTION__, orig);
				ccm_all_restart(hb, info, reply);
				break;
			}
			update_reset(CCM_GET_UPDATETABLE(info));
			update_add(CCM_GET_UPDATETABLE(info), CCM_GET_LLM(info),
						orig, uptime_val, TRUE);

			CCM_SET_MINORTRANS(info, trans_minorval);
			repeat = 0;
			while (ccm_send_join(hb, info) != HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}

			ccm_set_state(info, CCM_STATE_JOINING, reply);
			break;		

		case CCM_TYPE_STATE_INFO:
			ccm_handle_state_info(hb, info, reply);
			break;			
		case CCM_TYPE_RESTART:
			ccm_all_restart(hb, info, reply);
			break;

		default:  
			ccm_log(LOG_ERR, "%s: dropping message "
				"of type %s. Is this a Byzantine failure?", 
				__FUNCTION__, ccm_type2string(ccm_msg_type));
			/* nothing to do. Just forget the message */
			break;	
	}
}


/* */
/* The state machine that processes message when it is */
/*	in the CCM_STATE_SENT_MEMLISTREQ state */
/* */
static void
ccm_state_sent_memlistreq(enum ccm_type ccm_msg_type, 
			struct ha_msg *reply, 
			ll_cluster_t *hb, 
			ccm_info_t *info)
{
	const char *orig,  *trans, *memlist, *uptime;
	uint   trans_minorval=0, trans_majorval=0, trans_maxval=0;
        uint    uptime_val;
	int repeat;

	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown", __FUNCTION__);
		return;
	}

	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) { 
		ccm_debug(LOG_WARNING, "%s: received message from unknown host %s",
		__FUNCTION__, orig);
		return;
	}
	
	if(ccm_msg_type == CCM_TYPE_PROTOVERSION
	   || ccm_msg_type ==  CCM_TYPE_STATE_INFO
	   || ccm_msg_type == CCM_TYPE_RESTART) {		
		goto switchstatement;
	}

	if(strncmp(CCM_GET_COOKIE(info), ha_msg_value(reply, CCM_COOKIE), 
				COOKIESIZE) != 0){
		ccm_debug(LOG_WARNING, "%s: received message "
				"with unknown cookie, just dropping", __FUNCTION__);
		return;
	}

	/* get the major transition version */
	if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) { 
		ccm_debug(LOG_WARNING, "%s :no transition major information", __FUNCTION__);
		return;
	}

	trans_majorval = atoi(trans);
	 /*	drop the message if it has lower major transition number */
	if (CCM_TRANS_EARLIER(trans_majorval,  CCM_GET_MAJORTRANS(info))) {
		ccm_debug(LOG_WARNING, "%s: received CCM_TYPE_JOIN message with"
					"a earlier major transition number", __FUNCTION__);
		return;
	}


	/* get the minor transition version */
	if ((trans = ha_msg_value(reply, CCM_MINORTRANS)) == NULL) { 
		ccm_debug(LOG_WARNING, "%s: no transition minor information", __FUNCTION__);
		return;
	}

	trans_minorval = atoi(trans);

switchstatement:
	switch (ccm_msg_type)  {
		case CCM_TYPE_PROTOVERSION_RESP:

			ccm_debug(LOG_WARNING, "%s: "
				"dropping message of type %s. "
				" Is this a Byzantine failure?",
				__FUNCTION__, ccm_type2string(ccm_msg_type));

			break;


		case CCM_TYPE_PROTOVERSION:
			/*
			 * cache this request. We will respond to it, 
			 * if we become the leader.
			 */
		  ccm_add_new_joiner(info, orig, reply);
			
			break;

		case CCM_TYPE_JOIN:

			/* The join request has come too late.
			 * I am already the leader, and my
			 * leadership cannot be relinquished
			 * because that can confuse everybody.
			 * This join request shall be considered.
			 * But leadership shall not be relinquished.
			 */
			if(trans_majorval != CCM_GET_MAJORTRANS(info)
			|| trans_minorval != CCM_GET_MINORTRANS(info)) {
				ccm_log(LOG_WARNING,
				"%s: got a join message from %s from a wrong "
				"transition, restarting", __FUNCTION__, orig);
				ccm_all_restart(hb, info, reply);
				break;
			}
			ccm_debug2(LOG_DEBUG, "considering a late join message "
					  "from orig=%s", orig);
			/* get the update value */
			if ((uptime = ha_msg_value(reply, CCM_UPTIME)) 
						== NULL){
				ccm_debug(LOG_WARNING, 
					"%s: no update information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);
			update_add(CCM_GET_UPDATETABLE(info), 
				CCM_GET_LLM(info), orig, uptime_val, FALSE);
			ccm_memcomp_add(info, orig);
			break;

		case CCM_TYPE_TIMEOUT:
			if (ccm_memcomp_timeout(info,
				CCM_TMOUT_GET_IFF(info))) {
				/* we waited long for membership response 
				 * from all nodes, stop waiting and send
				 * final membership list
				 */				
				ccm_compute_and_send_final_memlist(hb, info);
			}
			break;

		case CCM_TYPE_REQ_MEMLIST:
			/* if this is my own message just forget it */
  			if(strncmp(orig, llm_get_mynodename(&info->llm),
  				   NODEIDSIZE) == 0){
				if(llm_get_live_nodecount(&info->llm) == 1){
					ccm_log(LOG_INFO, "%s: directly call"
						"ccm_compute_and_send_final_memlist()",
						__FUNCTION__);
					ccm_compute_and_send_final_memlist(hb, info);
				}
  				break;
  			}

			/* whoever is requesting memlist from me thinks it is 
			 * the leader. Hmm....., we will send it a NULL memlist.
			 * In partitioned network case both of us can be 
			 * leaders. Right?
			 */

			repeat = 0;
			while (ccm_send_memlist_res(hb, info, orig, NULL) != 
						HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_log(LOG_ERR,
						"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}
			break;

		case CCM_TYPE_RES_MEMLIST:
			/* mark that this node has sent us a memlist reply.
			 * Calculate the membership list with this new message 
			 */
			if(trans_minorval != CCM_GET_MINORTRANS(info)){
				break;
			}

			if(trans_majorval != CCM_GET_MAJORTRANS(info)) {
				ccm_log(LOG_WARNING, 
				   "%s: dropping CCM_TYPE_RES_MEMLIST "
				   "from orig=%s mymajor=%d msg_major=%d", 
				   __FUNCTION__, orig, trans_majorval, 
					CCM_GET_MAJORTRANS(info));
				break;
			}
			if ((memlist = ha_msg_value(reply, CCM_MEMLIST)) 
						== NULL) { 
				ccm_debug(LOG_WARNING, "%s: no memlist ", __FUNCTION__);
				break;
			}
			/* get the max transition version */
			if (!(trans = ha_msg_value(reply, CCM_MAXTRANS))) { 
				ccm_debug(LOG_WARNING, 
					"%s: no max transition "
					"information %s, type=%d", 
					__FUNCTION__, orig, ccm_msg_type);
				return;
			}

			trans_maxval = atoi(trans);

			ccm_memcomp_note(info, orig, trans_maxval, memlist);

			if (ccm_memcomp_rcvd_all(info)) {				
				ccm_compute_and_send_final_memlist(hb,info);
			}
			break;

		case CCM_TYPE_LEAVE: 
			/* since we are waiting for a memlist from all the 
			 * members who have sent me a join message, we 
			 * should be waiting for their message or their 
			 * leave message atleast.
			 */

			/* if this node had not participated in the update 
			 * exchange than just neglect it 
			 */
			if(!update_is_node_updated(CCM_GET_UPDATETABLE(info), 
						   CCM_GET_LLM(info), orig)) {
				break;
			}
			
			/* if this node had sent a memlist before dying,
			 * reset its memlist information */
			ccm_memcomp_note(info, orig, 0, "");

			if (ccm_memcomp_rcvd_all(info)) {
				ccm_compute_and_send_final_memlist(hb, info);
			}
			break;
			
		case CCM_TYPE_STATE_INFO:
			ccm_handle_state_info(hb, info, reply);
			break;
		case CCM_TYPE_RESTART:
			ccm_all_restart(hb, info, reply);
			break;

		case CCM_TYPE_FINAL_MEMLIST:
		case CCM_TYPE_ABORT:
		default:
			ccm_log(LOG_ERR, "%s: dropping message of type %s. Is this "
					"a Byzantine failure?", 
					__FUNCTION__, ccm_type2string(ccm_msg_type));
			/* nothing to do. Just forget the message */
			break;
	}
}

/* */
/* the state machine that processes messages when it is in the */
/* CCM_STATE_MEMLIST_RES state. */
/* */
static void
ccm_state_memlist_res(enum ccm_type ccm_msg_type, 
		struct ha_msg *reply, 
		ll_cluster_t *hb, 
		ccm_info_t *info)
{
	const char *orig,  *trans, *uptime, *memlist, *cookie, *cl;
	uint   trans_majorval=0, trans_minorval=0, trans_maxval=0;
	uint    uptime_val;
	uint  curr_major, curr_minor;
	int   indx;
	int repeat;
	int quorum;


	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown", __FUNCTION__);
		return;
	}

	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) { 
		ccm_debug(LOG_WARNING, "%s: received message from unknown host %s",
		__FUNCTION__, orig);
		return;
	}
	if(ccm_msg_type == CCM_TYPE_PROTOVERSION
	   || ccm_msg_type ==  CCM_TYPE_STATE_INFO
	   || ccm_msg_type == CCM_TYPE_RESTART) {
		goto switchstatement;
	}

	if(strncmp(CCM_GET_COOKIE(info), ha_msg_value(reply, CCM_COOKIE), 
				COOKIESIZE) != 0){
		ccm_debug(LOG_WARNING, "%s: received message with unknown cookie, just dropping",
		__FUNCTION__);
		return;
	}

	/* get the major transition version */
	if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) { 
		ccm_debug(LOG_WARNING, "%s: no transition major information", __FUNCTION__);
		return;
	}

	trans_majorval = atoi(trans);
	 /*	drop the message if it has lower major transition number */
	if (CCM_TRANS_EARLIER(trans_majorval,  CCM_GET_MAJORTRANS(info))) {
		ccm_debug(LOG_WARNING, "%s: received CCM_TYPE_JOIN message with"
					"a earlier major transition number", __FUNCTION__);
		return;
	}


	/* get the minor transition version */
	if ((trans = ha_msg_value(reply, CCM_MINORTRANS)) == NULL) { 
		ccm_debug(LOG_WARNING, "%s: no transition minor information", __FUNCTION__);
		return;
	}

	trans_minorval = atoi(trans);


switchstatement:

	switch (ccm_msg_type)  {
		case CCM_TYPE_PROTOVERSION_RESP:
			ccm_debug(LOG_WARNING, "%s: dropping message"
					" of type %s. Is this a Byzantine failure?", 
					__FUNCTION__, ccm_type2string(ccm_msg_type));
			break;

		case CCM_TYPE_PROTOVERSION:
			/*
			 * cache this request. We will respond to it, if we 
			 * become the leader.
			 */
		  ccm_add_new_joiner(info, orig, reply);
			
			break;

		case CCM_TYPE_JOIN:

			/*
			 * This could have happened because the leader died 
			 * and somebody noticed this and sent us this request. 
			 * In such a case the minor transition number should 
			 * have incremented. Or
			 * This could have happened because the leader's 
			 * FINAL_MEMLIST	
			 * has not reach us, whereas it has reached somebody 
			 * else, and since that somebody saw a change in 
			 * membership, initiated another join protocol. 
			 * In such a case the major transition
			 * number should have incremented.
			 */
			/* 
			 * if major number is incremented, send an abort message
			 * to the sender. The sender must resend the message.
			 */
			if (trans_majorval > CCM_GET_MAJORTRANS(info)) {
				repeat = 0;
				while (ccm_send_abort(hb, info, orig, 
					trans_majorval, trans_minorval) 
							!= HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_log(LOG_ERR,
						"%s: failure to send abort", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				break;
			}

			/* if minor transition number is incremented, 
			 * reset uptable table and start a join protocol
			 */
			if (trans_minorval > CCM_GET_MINORTRANS(info)) {
				/* get the update value */
				if ((uptime = ha_msg_value(reply, CCM_UPTIME)) 
							== NULL){
					ccm_debug(LOG_WARNING, 
						"%s: no update information", __FUNCTION__);
					return;
				}
				uptime_val = atoi(uptime);

				update_reset(CCM_GET_UPDATETABLE(info));
				update_add(CCM_GET_UPDATETABLE(info), 
					CCM_GET_LLM(info), orig, uptime_val, TRUE);

				CCM_SET_MINORTRANS(info, trans_minorval);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_log(LOG_ERR,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
			}

			break;
			

		case CCM_TYPE_REQ_MEMLIST:
			/* there are two reasons that can bring us here 
			 * 1. Because some other node still thinks he is 
			 * the master,(though we dont think so). Send 
			 * a NULL membership list to him immidiately.
			 * 2. Because of byzantine failures, though we have 
			 * not received the membership list in the last 
			 * round. We have waited to such an exent that some 
			 * node already thinks he is the master of the
			 * the new group transition. Well, there is something 
			 * seriously wrong with us. We will send a leave 
			 * message to everybody and say good bye. And we 
			 * will start all fresh!
			 */
			if (trans_minorval == CCM_GET_MINORTRANS(info)) {
				repeat = 0;
				while (ccm_send_memlist_res(hb, info, orig, 
							NULL) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_log(LOG_ERR,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				break;
			}

			break;

        	case CCM_TYPE_TIMEOUT:
			/* If we have waited too long for the leader to respond
			 * just assume that the leader is dead and start over
			 * a new round of the protocol
			 */
			if(!finallist_timeout(CCM_TMOUT_GET_FL(info))) {
				break;
			}
			update_reset(CCM_GET_UPDATETABLE(info));
			CCM_INCREMENT_MINORTRANS(info);
			repeat = 0;
			while (ccm_send_join(hb, info) != HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_log(LOG_ERR,
						"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}
			finallist_reset();
			ccm_set_state(info, CCM_STATE_JOINING, reply);
			break;

		case CCM_TYPE_LEAVE: 
			/* 
			 * If this message is because of loss of connectivity 
			 * with the node which we think is the master, then 
			 * restart the join. Loss of anyother node should be 
			 * confirmed by the finalmemlist of the master.
		 	 */
			cl = update_get_cl_name(CCM_GET_UPDATETABLE(info), 
					CCM_GET_LLM(info));
			if(strncmp(cl, orig, NODEIDSIZE) == 0) {
				/* increment the current minor transition value 
				 * and resend the join message 
				 */
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_log(LOG_ERR,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				finallist_reset();
				ccm_set_state(info, CCM_STATE_JOINING, reply);
			}

			break;
		
		case CCM_TYPE_FINAL_MEMLIST:
			/* WOW we received the membership list from the master.
			 * Check if I am part of the membership list. If not, 
			 * voluntarily leave the cluster and start all over 
			 * again 
			 */
			cl = update_get_cl_name(CCM_GET_UPDATETABLE(info), 
						CCM_GET_LLM(info));

			if(strncmp(cl, orig, NODEIDSIZE) != 0) {
				/* received memlist from a node we do not 
				 * think is the leader. We just reject the 
				 * message and wait for a message from the 
				 * our percieved master
				 */
				ccm_debug(LOG_WARNING, "%s: received final memlist from "
					"non-master,neglecting", __FUNCTION__);
									
				break;
			}
	
			/* 
			 * confirm that the major transition and minor 
			 * transition version match
			 */
			curr_major = CCM_GET_MAJORTRANS(info);
			curr_minor = CCM_GET_MINORTRANS(info);

			if(curr_major != trans_majorval || 
				curr_minor !=  trans_minorval){
				ccm_debug(LOG_WARNING, "%s: "
					"received final memlist from master, "
					"but transition versions do not match: "
					"rejecting the message", __FUNCTION__);
				break;
			}
			
			if ((memlist = ha_msg_value(reply, CCM_MEMLIST)) 
						== NULL) { 
				ccm_debug(LOG_WARNING, "%s: no membership list ", __FUNCTION__);
				return;
			}
			if ((trans = ha_msg_value(reply, CCM_MAXTRANS)) 
						== NULL) { 
				ccm_debug(LOG_WARNING, "%s: no membership list ", __FUNCTION__);
				return;
			}
			trans_maxval = atoi(trans);
			
			if (ha_msg_value_int (reply, CCM_QUORUM, &quorum)==HA_OK){ 
				info->has_quorum = quorum;
			}
			else {
				info->has_quorum = -1;
			}

			if (!am_i_member_in_memlist(info, memlist)) {
				ccm_reset(info); 
				break;
			}

			ccm_mem_strfill(info, (const char *)memlist);
			/* increment the major transition number and reset the
			 * minor transition number
			 */
			CCM_SET_MAJORTRANS(info, trans_maxval); 
			CCM_RESET_MINORTRANS(info);

			/* check if leader has changed the COOKIE, this can
			 * happen if the leader sees a partitioned group
			 */
			if ((cookie = ha_msg_value(reply, CCM_NEWCOOKIE)) 
						!= NULL) { 
				ccm_debug2(LOG_DEBUG, "%s: leader  changed  cookie ", __FUNCTION__);
				CCM_SET_COOKIE(info, cookie); 
			}

			indx = llm_get_index(&info->llm, cl); 
			assert(indx != -1);
			CCM_SET_CL(info, indx); 
			report_mbrs(info); /* call before update_reset */
/*			update_reset(CCM_GET_UPDATETABLE(info));*/
			finallist_reset();
			ccm_set_state(info, CCM_STATE_JOINED, reply);
			ccm_reset_all_join_request(info);
			if(!ccm_already_joined(info)) 
				CCM_SET_JOINED_TRANSITION(info, 
					CCM_GET_MAJORTRANS(info));
			break;

		case CCM_TYPE_STATE_INFO:
			ccm_handle_state_info(hb, info, reply);
			break;		

		case CCM_TYPE_RESTART:
			ccm_all_restart(hb, info, reply);
			break;

		case CCM_TYPE_ABORT:
		case CCM_TYPE_RES_MEMLIST:
		default:
			ccm_log(LOG_ERR, "%s: dropping message of type %s. "
					"Is this a Byzantine failure?", 
					__FUNCTION__, ccm_type2string(ccm_msg_type));
			/* nothing to do. Just forget the message */
			break;
	}
}



/* */
/* the state machine that processes messages when it is in the */
/* CCM_STATE_JOINING state. */
/* */
static void
ccm_state_joining(enum ccm_type ccm_msg_type, 
		struct ha_msg *reply, 
		ll_cluster_t *hb, 
		ccm_info_t *info)
{
	const char *orig,  *trans, *uptime;
	uint   trans_majorval=0, trans_minorval=0;
        uint	uptime_val;
	int repeat;
	
	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown", __FUNCTION__);
		return;
	}

	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) { 
		ccm_debug(LOG_WARNING, "%s: received message "
				"from unknown host %s", __FUNCTION__, orig);
		return;
	}

	if(ccm_msg_type == CCM_TYPE_PROTOVERSION
	   || ccm_msg_type ==  CCM_TYPE_STATE_INFO
	   || ccm_msg_type == CCM_TYPE_RESTART) {
		goto switchstatement;
	}

	if(strncmp(CCM_GET_COOKIE(info), ha_msg_value(reply, CCM_COOKIE), 
			COOKIESIZE) != 0){

		if(ccm_msg_type ==  CCM_TYPE_PROTOVERSION_RESP) {
			version_inc_nresp(CCM_GET_VERSION(info));
			ccm_debug(LOG_WARNING, "%s: received message "
			"incrementing versionresp counter %d", 
				__FUNCTION__, version_get_nresp(CCM_GET_VERSION(info)));
		}

		ccm_debug(LOG_WARNING, "%s: received message "
			"with unknown cookie, just dropping", __FUNCTION__);
		return;
	}

	


	/* get the major transition version */
	if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) { 
		ccm_debug(LOG_WARNING, "%s: no transition major information", __FUNCTION__);
		return;
	}

	trans_majorval = atoi(trans);
	 /*	drop the message if it has lower major transition number */
	if (CCM_TRANS_EARLIER(trans_majorval,  CCM_GET_MAJORTRANS(info))) {
		ccm_debug(LOG_WARNING, "%s: received CCM_TYPE_JOIN message with"
				"a earlier major transition number", __FUNCTION__);
		return;
	}


	/* get the minor transition version */
	if ((trans = ha_msg_value(reply, CCM_MINORTRANS)) == NULL) { 
		ccm_debug(LOG_WARNING, "%s: no transition minor information", __FUNCTION__);
		return;
	}

	trans_minorval = atoi(trans);
	if (trans_minorval < CCM_GET_MINORTRANS(info)) {
		return;
	}


switchstatement:
	switch (ccm_msg_type)  {

		case CCM_TYPE_PROTOVERSION_RESP:

			/* If we were joined in an earlier iteration, then this
			 * message should not have arrived. A bug in the logic!
			 */
			if(ccm_already_joined(info)) {
 				ccm_debug(LOG_WARNING, "%s: BUG:"
					" received CCM_TYPE_PROTOVERSION_RESP "
					"message when we have not asked for "
 					"it ", __FUNCTION__);
				break;
			}

  			ccm_debug(LOG_WARNING, "%s: dropping message "
 					" of type %s. Is this a Byzantine failure?", 
 					__FUNCTION__, ccm_type2string(ccm_msg_type));
			break;
				

		case CCM_TYPE_PROTOVERSION:
			/*
			 * cache this request. We will respond to it, 
			 * if we become the leader.
			 */
		  ccm_add_new_joiner(info, orig, reply);
			
			break;

        	case CCM_TYPE_JOIN:
			/* get the update value */
			if((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){ 
				ccm_debug(LOG_WARNING, "%s: no update information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);

			/* 
			 * note down all the information contained in the 
			 * message There is a possibility that I am the leader,
			 * if all the nodes died, and I am the only surviving 
			 * node! If this message has originated from me, 
			 * note down the current time. This information is 
			 * needed, to later recognize that I am the only 
			 * surviving node.
			 */
			/* update the minor transition number if it is of 
			 * higher value 
			 * and send a fresh JOIN message 
			 */
			if (trans_minorval > CCM_GET_MINORTRANS(info)) {
				update_reset(CCM_GET_UPDATETABLE(info));
				update_add( CCM_GET_UPDATETABLE(info),
					CCM_GET_LLM(info), orig, uptime_val, TRUE);

				CCM_SET_MINORTRANS(info, trans_minorval);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_log(LOG_ERR, 
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
			} else {
				/* update the update table  */
				update_add( CCM_GET_UPDATETABLE(info),
					CCM_GET_LLM(info), orig, uptime_val, 
					TRUE);

				/* if all nodes have responded, its time 
				 * to elect the leader 
				 */
				if (UPDATE_GET_NODECOUNT(
					CCM_GET_UPDATETABLE(info)) ==
				    llm_get_live_nodecount(&info->llm)) {

					/* check if I am the leader */
					if (update_am_i_leader(
						CCM_GET_UPDATETABLE(info),
						CCM_GET_LLM(info))) {
						/* send out the 
						 * membershiplist request */
						repeat = 0;
						while(ccm_send_memlist_request(
							hb, info)!=HA_OK) {
							if(repeat < REPEAT_TIMES){
							ccm_log(LOG_ERR, 
							"%s: failure to send memlist request",
							__FUNCTION__);
							cl_shortsleep();
							repeat++;
							}else{
								break;
							}
						}
						ccm_memcomp_init(info);
						ccm_memcomp_note_my_membership(
								info);
						ccm_set_state(info, 
							      CCM_STATE_SENT_MEMLISTREQ, reply);
					} else {
						/* check if we have already 
						 * received memlist request
						 * from any node(which 
						 * believes itself to be the 
						 * leader)
						 * If so,we have to reply to 
						 * them with our membership
						 * list. But there is a catch. 
						 * If we do not think the
						 * requestor to be the leader, 
						 * then we send it an null
						 * membership message!
						 */
						if (ccm_send_cl_reply(hb,info) 
								== TRUE) {
							finallist_init();
							ccm_set_state(info, 
								      CCM_STATE_MEMLIST_RES, reply);
						}
					}
					break; /* done all processing */
				} 
			}
				   
			break;	

		case CCM_TYPE_REQ_MEMLIST:

			/* well we have not yet timedout! And a memlist
			 * request has arrived from the cluster leader.  Hmm...
			 * We should wait till timeout, to respond.
			 *
			 * NOTE:  there is a chance
			 * that more than one cluster leader might request
			 * the membership list. Due to cluster partitioning :( )
			 */
			/* If we have received CCM_TYPE_JOIN from all nodes
			 * we don't need wait for timeout here.
			 */
			update_add_memlist_request(CCM_GET_UPDATETABLE(info),
					CCM_GET_LLM(info), orig, trans_majorval);


			if (UPDATE_GET_NODECOUNT( CCM_GET_UPDATETABLE(info)) == llm_get_live_nodecount(&info->llm)
			    && !update_am_i_leader(CCM_GET_UPDATETABLE(info), CCM_GET_LLM(info))) {	
				
				if (ccm_send_cl_reply(hb,info) == TRUE) {
					finallist_init();
					ccm_set_state(info, CCM_STATE_MEMLIST_RES, reply);
					break;
				}					
			}
			
			/*
			 * FALL THROUGH
			 */
		case CCM_TYPE_TIMEOUT:
			/*
			 * If timeout expired, elect the leader.
			 * If I am the leader, send out the membershiplist request
			 */
			if (!update_timeout_expired(CCM_GET_UPDATETABLE(info), 
					CCM_TMOUT_GET_U(info))) {
						break;
			}
			

			if (update_am_i_leader(CCM_GET_UPDATETABLE(info),
						CCM_GET_LLM(info))) {

				/* if I am the only one around go directly
				 * to joined state.
				 */
				if (UPDATE_GET_NODECOUNT(
					CCM_GET_UPDATETABLE(info)) == 1) {

					if(ccm_already_joined(info) || 
						!version_get_nresp(
						  CCM_GET_VERSION(info))){
						ccm_joining_to_joined(hb,
							       	info);
					} else {
						ccm_reset(info);
					}
					break;
				}

				/* send out the membershiplist request */
				repeat = 0;
				while (ccm_send_memlist_request(hb, info) 
							!= HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_log(LOG_ERR,
						"%s: failure to send memlist request",
						__FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_memcomp_init(info);
				ccm_memcomp_note_my_membership(info);
				ccm_set_state(info, CCM_STATE_SENT_MEMLISTREQ, reply);
			} else {
				/* check if we have already received memlist 
				 * request from any node(which believes itself 
				 * to be the leader)
				 * If so,we have to reply to them with our 
				 * membership list. But there is a catch. 
				 * If we do not think the
				 * requestor to be the leader, then we send 
				 * it an abort message!
				 */
				if (ccm_send_cl_reply(hb, info) == TRUE) {
					/* free the update data*/
					finallist_init();
					ccm_set_state(info, 
						      CCM_STATE_MEMLIST_RES, reply);
				}
			}
			break;


		case CCM_TYPE_ABORT:

			/*
			 * This is a case where my JOIN request is not honoured
			 * by the recieving host(probably because it is waiting
			 * on some message, before which it cannot initiate 
			 * the join).
			 * We will resend the join message, incrementing the
			 * minor version number, provided this abort is 
			 * requested
			 * for this minor version.
			 */
			if(trans_majorval != CCM_GET_MAJORTRANS(info) ||
				trans_minorval != CCM_GET_MINORTRANS(info)) {
				/* nothing to worry  just forget this message */
					break;
			}
					
			/* increment the current minor transition value 
			 * and resend the
				join message */
			CCM_INCREMENT_MINORTRANS(info);
			update_reset(CCM_GET_UPDATETABLE(info));
			repeat = 0;
			while (ccm_send_join(hb, info) != HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_log(LOG_ERR,
						"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}

			break;

		case CCM_TYPE_LEAVE: 

			/* 
			 * Has that node already sent a valid update message 
			 * before death. If so, remove him from the update 
			 * table.
			 */
			update_remove(CCM_GET_UPDATETABLE(info),
					CCM_GET_LLM(info), 
					orig);
			/* if we have any cached version-request from this node 
			 * we will get rid of that too
			 */
			ccm_remove_new_joiner(info, orig);
			break;
			
		case CCM_TYPE_STATE_INFO:
			ccm_handle_state_info(hb, info, reply);
			break;
		case CCM_TYPE_RESTART:
			ccm_all_restart(hb, info, reply);
			break;

		case CCM_TYPE_RES_MEMLIST:
		case CCM_TYPE_FINAL_MEMLIST:
			/* this message is from other partitions*/
			ccm_debug(LOG_WARNING, "%s: received a %s message", 
			       __FUNCTION__, ccm_type2string(ccm_msg_type));
			ccm_debug(LOG_WARNING, "We probably have different partitions");
			break;
			
			
		default:
			ccm_log(LOG_ERR, "%s: dropping message "
			       "of type %s from %s. Is this a Byzantine failure?", 
			       __FUNCTION__, ccm_type2string(ccm_msg_type), orig);
			/* nothing to do. Just forget the message */
			break;
	}
	return;
}


static void
ccm_control_init(ccm_info_t *info)
{
	ccm_init(info);
	
	/* if this is the only active node in the cluster, go to the 
	   JOINED state */
	if (llm_get_live_nodecount(CCM_GET_LLM(info)) == 1) {
		ccm_init_to_joined(info);
	} else {
		ccm_set_state(info, CCM_STATE_NONE, NULL);
	}
	
	return;
}



/* */
/* The callback function which is called when the status of a link */
/* changes. */
/* */
static void
LinkStatus(const char * node, const char * lnk, const char * status ,
		void * private)
{
	ccm_debug2(LOG_DEBUG, "Link Status update: Link %s/%s "
			"now has status %s", node, lnk, status);
}





/*  */
/* The most important function which tracks the state machine. */
/*  */




/*  look at the current state machine and decide if  */
/*  the state machine needs immidiate control for further */
/*  state machine processing. Called by the check function */
/*  of heartbeat-source of the main event loop. */
int
ccm_need_control(void *data)
{
	ccm_info_t *info =  (ccm_info_t *)((ccm_t *)data)->info;

	if(leave_any() || 
	   CCM_GET_STATE(info) != CCM_STATE_JOINED){
			return TRUE;
	}
	return FALSE;
}

/*  look at the current state machine and decide if  */
/*  the state machine needs immidiate control for further */
/*  state machine processing. Called by the check function */
/*  of heartbeat-source of the main event loop. */
int
ccm_take_control(void *data)
{
	ccm_info_t *info =  (ccm_info_t *)((ccm_t *)data)->info;
	ll_cluster_t *hbfd = (ll_cluster_t *)((ccm_t *)data)->hbfd;

	return  ccm_control_process(info, hbfd);
}

IPC_Channel *
ccm_get_ipcchan(void *data)
{
	ll_cluster_t *hbfd = (ll_cluster_t *)((ccm_t *)data)->hbfd;

	return hbfd->llc_ops->ipcchan(hbfd);
}


#define	PINGNODE        "ping"
static int
set_llm_from_heartbeat(ll_cluster_t* llc, ccm_info_t* info){
	llm_info_t*	llm = &info->llm;
	struct llc_ops* ops = llc->llc_ops;
	const char*	status;
	const char*	node;
	const char*	mynode = ops->get_mynodeid(llc);
	const char*	cluster;
	const char*	quorum_server;
	const char*	site;
	int		weight;
	
	if (mynode == NULL){
		ccm_log(LOG_ERR, "%s: mynode is NULL", __FUNCTION__);
		return HA_FAIL;
	}

	ccm_debug2(LOG_DEBUG, "==== Starting  Node Walk =========");
	
	if (ops->init_nodewalk(llc) != HA_OK) {
		ccm_log(LOG_ERR, "Cannot start node walk");
		ccm_log(LOG_ERR, "REASON: %s", ops->errmsg(llc));
		return HA_FAIL;
	}
	
	llm = CCM_GET_LLM(info);
	llm_init(llm);
	memset(info->cluster, 0, sizeof(info->cluster));
	cluster = llc->llc_ops->get_parameter(llc, KEY_CLUSTER);
	if (cluster != NULL) {
		strncpy(info->cluster, cluster, PATH_MAX);
	}
	memset(info->quorum_server, 0, sizeof(info->quorum_server));
	quorum_server = llc->llc_ops->get_parameter(llc, KEY_QSERVER);
	if (quorum_server != NULL) {
		strncpy(info->quorum_server, quorum_server, PATH_MAX);
	}
	

	while((node = ops->nextnode(llc)) != NULL) {		
		if (strcmp(ops->node_type(llc, node), PINGNODE)==0){
			continue;
		}
		
		status = ops->node_status(llc, node);
		site = ops->node_site(llc, node);
		weight = ops->node_weight(llc, node);
		ccm_debug2(LOG_DEBUG, "Cluster node: %s: status: %s", node,
			       status);
		
		if (llm_add(llm, node, status, mynode, site, weight)!= HA_OK){
			ccm_log(LOG_ERR, "%s: adding node %s to llm failed",
			       __FUNCTION__, node);
			return HA_FAIL;
			
		}
		
	}
	
	llm_display(llm);
	
	if (ops->end_nodewalk(llc) != HA_OK) {
		ccm_log(LOG_ERR, "Cannot end node walk");
		ccm_log(LOG_ERR, "REASON: %s", ops->errmsg(llc));
		return HA_FAIL;
	}
	
	ccm_debug2(LOG_DEBUG, "======= Ending  Node Walk ==========");
	ccm_debug2(LOG_DEBUG, "Total # of Nodes in the Cluster: %d", 
		       llm_get_nodecount(llm));
	
	return HA_OK;
}

ccm_info_t* ccm_info_saved = NULL;
ll_cluster_t* hb_fd_saved = NULL;

void *
ccm_initialize()
{
	unsigned	fmask;
	const char *	hname;
	ccm_info_t 	*global_info = NULL;
	ll_cluster_t*	hb_fd;
	ccm_t		*ccmret = NULL;
	const char *	parameter;

	ccm_debug2(LOG_DEBUG, "========================== Starting CCM ===="
			"======================");

	CL_SIGINTERRUPT(SIGTERM, 1);
	cl_inherit_logging_environment(0);
	
	hb_fd = ll_cluster_new("heartbeat");

	ccm_debug(LOG_DEBUG, "Signing in with Heartbeat");
	if (hb_fd->llc_ops->signon(hb_fd, "ccm")!= HA_OK) {
		ccm_log(LOG_ERR, "Cannot sign on with heartbeat");
		ccm_log(LOG_ERR, "REASON: %s", hb_fd->llc_ops->errmsg(hb_fd));
		goto errout;
	}

	/* See if we should drop cores somewhere odd... */
	parameter = hb_fd->llc_ops->get_parameter(hb_fd, KEY_COREROOTDIR);
	if (parameter) {
		cl_set_corerootdir(parameter);
	}
	cl_cdtocoredir();

	if((global_info = (ccm_info_t *)g_malloc(sizeof(ccm_info_t))) == NULL){
		ccm_log(LOG_ERR, "Cannot allocate memory ");
		goto errout;
	}

	memset(global_info, 0, sizeof(ccm_info_t));

	if((ccmret = (ccm_t *)g_malloc(sizeof(ccm_t))) == NULL){
		ccm_log(LOG_ERR, "Cannot allocate memory");
		goto errout;
	}

	if((hname = hb_fd->llc_ops->get_mynodeid(hb_fd)) == NULL) {
		ccm_log(LOG_ERR, "get_mynodeid() failed");
		goto errout;
	}
	ccm_log(LOG_INFO, "Hostname: %s", hname);


	if (hb_fd->llc_ops->set_ifstatus_callback(hb_fd, LinkStatus, NULL)
	    !=HA_OK){
		ccm_log(LOG_ERR, "Cannot set if status callback");
		ccm_log(LOG_ERR, "REASON: %s", hb_fd->llc_ops->errmsg(hb_fd));
		goto errout;
	}
	
	fmask = LLC_FILTER_DEFAULT;
	if (hb_fd->llc_ops->setfmode(hb_fd, fmask) != HA_OK) {
		ccm_log(LOG_ERR, "Cannot set filter mode");
		ccm_log(LOG_ERR, "REASON: %s", hb_fd->llc_ops->errmsg(hb_fd));
		goto errout;
	}

	/* we'll benefit from a bigger queue length on heartbeat side.
	* Otherwise, if peers send messages faster than we can consume
	* them right now, heartbeat messaging layer will kick us out once
	* it's (small) default queue fills up :(
	* If we fail to adjust the sendq length, that's not yet fatal, though.
	*/
	if (HA_OK != hb_fd->llc_ops->set_sendq_len(hb_fd, 1024)) {
		ccm_log(LOG_WARNING, "Cannot set sendq length: %s",
			hb_fd->llc_ops->errmsg(hb_fd));
	}

	if (set_llm_from_heartbeat(hb_fd, global_info) != HA_OK){
		goto errout;
	}

	ccm_control_init(global_info);
	ccm_configure_timeout(hb_fd, global_info);

	ccmret->info = global_info;
	ccmret->hbfd = hb_fd;
	client_llm_init(&global_info->llm);
	ccm_info_saved = global_info;
	hb_fd_saved = hb_fd;
	return  (void*)ccmret;

 errout:
	if (ccmret){
		g_free(ccmret);
		ccmret = NULL;
	}
	if (global_info){
		g_free(global_info);
		global_info = NULL;
	}	
	return NULL;


}

static void add_change_msg(ccm_info_t *info, const char *node, const char *orig, enum change_event_type type)
{
	strlcpy(info->change_node_id, node, sizeof(info->change_node_id));
	info->change_type = type;
	if(type == NODE_LEAVE){
		info->change_event_remaining_count = ccm_get_memcount(info)-1;
	}else{
		info->change_event_remaining_count = ccm_get_memcount(info);
	}
	append_change_msg(info, orig);
	return;
}

static void append_change_msg(ccm_info_t *info, const char *node)
{
	if (CCM_GET_RECEIVED_CHANGE_MSG(info, node) == 0){
		CCM_SET_RECEIVED_CHANGE_MSG(info, node, 1);
		info->change_event_remaining_count--;
	}
	return;
}

static int received_all_change_msg(ccm_info_t *info)
{
	if(info->change_event_remaining_count == 0){
		return 1;
	}else{
		return 0;
	}
}

static int is_expected_change_msg(ccm_info_t *info, const char *node,enum change_event_type type)
{
	if(strcmp(info->change_node_id, node) == 0){
		if(info->change_type == type){
			return 1;
		}
	}
	return 0;
}


static void ccm_state_wait_for_mem_list(enum ccm_type ccm_msg_type, 
			struct ha_msg *reply, 
			ll_cluster_t *hb, 
			ccm_info_t *info)
{
	const char *orig, *trans, *uptime, *cookie, *memlist;
	int uptime_list[MAXNODE];
	size_t uptime_size = MAXNODE;
	uint trans_majorval=0,trans_minorval=0, uptime_val;
	uint curr_major, curr_minor;
	int repeat;
	int quorum;
	
	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown", __FUNCTION__);
		return;
	}

	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown host %s",
					__FUNCTION__, orig);
		return;
	}

	
	if(ccm_msg_type != CCM_TYPE_PROTOVERSION
	   && ccm_msg_type !=  CCM_TYPE_STATE_INFO
	   && ccm_msg_type != CCM_TYPE_RESTART) {
		if(strncmp(CCM_GET_COOKIE(info),
			ha_msg_value(reply, CCM_COOKIE), COOKIESIZE) != 0){
			ccm_debug(LOG_WARNING, "%s: received message"
					" with unknown cookie, just dropping", __FUNCTION__);
			return;
		}

		/* get the major transition version */
		if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) {
			ccm_debug(LOG_WARNING, "%s: no transition major information", __FUNCTION__);
			return;
		}
		trans_majorval = atoi(trans);

		/* drop the message if it has lower major transition number */
		if (CCM_TRANS_EARLIER(trans_majorval,  
					CCM_GET_MAJORTRANS(info))) {
			ccm_debug(LOG_WARNING, "%s: received %s message with "
				"a earlier major transition number "
				"recv_trans=%d, mytrans=%d",
				__FUNCTION__, ccm_type2string(ccm_msg_type), trans_majorval, 
				CCM_GET_MAJORTRANS(info));
			return;
		}

		/* get the minor transition version */
		if ((trans = ha_msg_value(reply, CCM_MINORTRANS)) == NULL) { 
			ccm_debug(LOG_WARNING, "%s: no transition minor information", __FUNCTION__);
			return;
		}

		trans_minorval = atoi(trans);
	}


	switch(ccm_msg_type){
		
		case CCM_TYPE_MEM_LIST:
        	
			curr_major = CCM_GET_MAJORTRANS(info);
			curr_minor = CCM_GET_MINORTRANS(info);

			if(curr_major != trans_majorval || 
				curr_minor !=  trans_minorval){
				ccm_debug(LOG_WARNING, "%s: "
					"received final memlist from master, "
					"but transition versions do not match: "
					"rejecting the message", __FUNCTION__);
				break;
			}
			if ((memlist = ha_msg_value(reply, CCM_MEMLIST)) 
							== NULL) {
				ccm_debug(LOG_WARNING, "%s: no membership list ", __FUNCTION__);
				return;
			}

			
			if (cl_msg_get_list_int(reply,CCM_UPTIMELIST, 
						uptime_list, &uptime_size) != HA_OK){
				ccm_log(LOG_ERR,"%s: getting uptie_list failed",
					__FUNCTION__);
				return;
			}
			
			if (ha_msg_value_int (reply, CCM_QUORUM, &quorum)==HA_OK){ 
				info->has_quorum = quorum;
			}
			else {
				info->has_quorum = -1;
			}

			ccm_mem_strfill(info, (const char *)memlist);
			CCM_SET_MAJORTRANS(info, curr_major+1);
			CCM_RESET_MINORTRANS(info);
			if ((cookie = ha_msg_value(reply, CCM_NEWCOOKIE))
						!= NULL) { 
				ccm_debug2(LOG_DEBUG, "%s: leader  changed  cookie ", __FUNCTION__);
				CCM_SET_COOKIE(info, cookie);
			}
			CCM_SET_CL(info, llm_get_index(&info->llm,orig));
			ccm_fill_update_table(info, CCM_GET_UPDATETABLE(info),
						uptime_list);
			report_mbrs(info);
			ccm_set_state(info, CCM_STATE_JOINED, reply);
			break;
        	
		case CCM_TYPE_TIMEOUT:
        		if (mem_list_timeout(CCM_TMOUT_GET_U(info))){
				reset_change_info(info);
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
			}
			break;

		case CCM_TYPE_JOIN:
        		/* get the update value */
			if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
				ccm_debug(LOG_WARNING, "%s: no update information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);

			/* update the minor transition number if it is of
			 * higher value and send a fresh JOIN message
			 */
			if (trans_minorval < CCM_GET_MINORTRANS(info)) {
				ccm_log(LOG_WARNING,
				"%s: got a join message from %s from earlier "
				"transition, restarting", __FUNCTION__, orig);
				ccm_all_restart(hb, info, reply);
				break;
			}
			update_reset(CCM_GET_UPDATETABLE(info));
			update_add(CCM_GET_UPDATETABLE(info), CCM_GET_LLM(info),
						orig, uptime_val, TRUE);

			CCM_SET_MINORTRANS(info, trans_minorval);
			repeat = 0;
			while (ccm_send_join(hb, info) != HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}

			ccm_set_state(info, CCM_STATE_JOINING, reply);
			break;

		case CCM_TYPE_LEAVE:
			
			/* if the dead node is leader, jump to CCM state machine */
			if(node_is_leader(info, orig)){
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
				return;
			}

		case CCM_TYPE_ALIVE:
			/* We do nothing here because we believe leader
			 * will deal with this LEAVE message. SPOF?
			 */
			break;

		case CCM_TYPE_PROTOVERSION:
			/* leader will handle this message
			 * we can safely ignore it
			 */
			break;


		case CCM_TYPE_STATE_INFO:
			ccm_handle_state_info(hb, info, reply);
			break;		
		case CCM_TYPE_RESTART:
			ccm_all_restart(hb, info, reply);
			break;


	
		default:
			ccm_log(LOG_ERR, "%s: dropping message "
				"of type %s. Is this a Byzantine failure?",
				__FUNCTION__, ccm_type2string(ccm_msg_type));
			/* nothing to do. Just forget the message */
			break;
	}
}


static void
reset_change_info(ccm_info_t *info)
{
	llm_info_t *llm = CCM_GET_LLM(info);
	unsigned i;

	for(i=0; i<llm_get_nodecount(llm); i++) {
		llm_set_change(llm, i, FALSE);
	}
	return;
}


/* */
/*  Construct and send mem_list, uptime_list to all members in the partition */
/* */
static void send_mem_list_to_all(ll_cluster_t *hb, 
		ccm_info_t *info, char *cookie)
{
	int i, size, j, tmp, tmp_mem[100];
	char *bitmap;
	char memlist[MAX_MEMLIST_STRING];
	int uptime[MAXNODE];
    
	bitmap_create(&bitmap, MAXNODE);
	size = info->memcount;
	
	for (i=0; i<size; i++){
		tmp_mem[i] = info->ccm_member[i]; 
	}	
        for (i=0; i<size; i++){
                for(j=0; j<(size-1-i); j++){
                        if(tmp_mem[j] > tmp_mem[j+1]){
                                tmp = tmp_mem[j];
                                tmp_mem[j] = tmp_mem[j+1];
                                tmp_mem[j+1] = tmp;
                        }
                }
        }

	for ( i = 0 ; i < size ; i++ ) {
		bitmap_mark(info->ccm_member[i], 
			    bitmap, MAXNODE);
		uptime[i] = htonl(update_get_uptime(CCM_GET_UPDATETABLE(info), 
						    CCM_GET_LLM(info),
						    tmp_mem[i]));
	}    
	ccm_bitmap2str(bitmap, memlist, MAX_MEMLIST_STRING);
	bitmap_delete(bitmap);
	ccm_send_to_all(hb, info, memlist, cookie, uptime, size);
	return;
}

static void ccm_state_new_node_wait_for_mem_list(enum ccm_type ccm_msg_type, 
	              struct ha_msg *reply, 
	              ll_cluster_t *hb, 
			ccm_info_t *info)
{
    	const char *orig,  *trans, *uptime, *memlist, *cookie;
	int uptime_list[MAXNODE];
	size_t uptime_size = MAXNODE;
	uint  trans_majorval=0,trans_minorval=0, uptime_val;
	uint  curr_major, curr_minor;
	int repeat;
	int ret;
	int quorum;

	if ((orig = ha_msg_value(reply, F_ORIG)) == NULL) {
		ccm_debug(LOG_WARNING, "%s: received message from unknown", __FUNCTION__);
		return;
	}

	if(!llm_is_valid_node(CCM_GET_LLM(info), orig)) { 
		ccm_debug(LOG_WARNING, "%s: received message from unknown host %s", __FUNCTION__, orig);
		return;
	}

	if(ccm_msg_type != CCM_TYPE_PROTOVERSION
	   && ccm_msg_type !=  CCM_TYPE_STATE_INFO
	   && ccm_msg_type != CCM_TYPE_RESTART) {
		
		if(strncmp(CCM_GET_COOKIE(info), 
			ha_msg_value(reply, CCM_COOKIE), COOKIESIZE) != 0){
			ccm_debug(LOG_WARNING, "%s: received message with unknown cookie, just dropping", __FUNCTION__);
			return;
		}

		/* get the major transition version */
		if ((trans = ha_msg_value(reply, CCM_MAJORTRANS)) == NULL) { 
			ccm_debug(LOG_WARNING, "%s: no transition major information", __FUNCTION__);
			return;
		}
		trans_majorval = atoi(trans);

	 	/*drop the message if it has lower major transition number */
		if (CCM_TRANS_EARLIER(trans_majorval,  
					CCM_GET_MAJORTRANS(info))) {
			ccm_debug(LOG_WARNING, "%s: received"
				" %s message with "
				"a earlier major transition number "
				"recv_trans=%d, mytrans=%d",
				__FUNCTION__, ccm_type2string(ccm_msg_type), trans_majorval, 
				CCM_GET_MAJORTRANS(info));
			return;
		}

		/* get the minor transition version */
		if ((trans = ha_msg_value(reply, CCM_MINORTRANS)) == NULL) { 
			ccm_debug(LOG_WARNING, "%s: no transition minor information", __FUNCTION__);
			return;
		}

		trans_minorval = atoi(trans);
	}
    	
	switch(ccm_msg_type){
		
		case CCM_TYPE_MEM_LIST:
			curr_major = CCM_GET_MAJORTRANS(info);
			curr_minor = CCM_GET_MINORTRANS(info);

			if(curr_major != trans_majorval || 
				curr_minor !=  trans_minorval){
				ccm_debug(LOG_WARNING, "%s: received final memlist from master, "
					"but transition versions do not match: "
					"rejecting the message", __FUNCTION__);
				break;
			}
			if ((memlist = ha_msg_value(reply, CCM_MEMLIST)) 
						== NULL) { 
				ccm_debug(LOG_WARNING, "%s: no membership list ", __FUNCTION__);
				return;
			}
			
			if (cl_msg_get_list_int(reply,CCM_UPTIMELIST, 
						uptime_list, &uptime_size) != HA_OK){
				ccm_log(LOG_ERR,"%s: getting uptie_list failed", __FUNCTION__);
				return;
			}
			
			ret = ccm_mem_strfill(info, (const char *)memlist);
			if (ret != HA_OK){
				ccm_log(LOG_ERR, "%s: filling membership from string failed",
				       __FUNCTION__);
				return;
			}
			
			ret = ccm_mem_filluptime(info, uptime_list, uptime_size);
			if (ret != HA_OK){
				ccm_log(LOG_ERR, "%s: filling uptime failed",
				       __FUNCTION__);
				return;
			}
			
			if (ha_msg_value_int (reply, CCM_QUORUM, &quorum)==HA_OK){ 
				info->has_quorum = quorum;
			}
			else {
				info->has_quorum = -1;
			}
			
			if (i_am_member(info) == FALSE){
				version_reset(CCM_GET_VERSION(info));
				ccm_set_state(info, CCM_STATE_NONE, reply);
				ccm_reset_all_join_request(info);
				break;
			}
			
			
			
			CCM_SET_MAJORTRANS(info, curr_major+1); 
			CCM_RESET_MINORTRANS(info);
			if ((cookie = ha_msg_value(reply, CCM_NEWCOOKIE)) 
						!= NULL) { 
				ccm_debug2(LOG_DEBUG, "%s: leader  changed  cookie ", __FUNCTION__);
				CCM_SET_COOKIE(info, cookie); 
			}
			CCM_SET_CL(info,llm_get_index(&info->llm, orig));
			CCM_SET_JOINED_TRANSITION(info, CCM_GET_MAJORTRANS(info));
			ccm_fill_update_table(info, 
				CCM_GET_UPDATETABLE(info), uptime_list);
			ccm_set_state(info, CCM_STATE_JOINED, reply);	        
			report_mbrs(info);
			break;

		case CCM_TYPE_TIMEOUT:
			if (new_node_mem_list_timeout(CCM_TMOUT_GET_U(info))){
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
			}	
			break;

		case CCM_TYPE_JOIN:
			/* get the update value */
			if ((uptime = ha_msg_value(reply, CCM_UPTIME)) == NULL){
				ccm_debug(LOG_WARNING, "%s: no update information", __FUNCTION__);
				return;
			}
			uptime_val = atoi(uptime);

			/* update the minor transition number if it is of 
			 * higher value and send a fresh JOIN message 
			 */
			if (trans_minorval < CCM_GET_MINORTRANS(info)) {
				ccm_log(LOG_WARNING,
				"%s: got a join message from %s from earlier "
				"transition, restarting", __FUNCTION__, orig);
				ccm_all_restart(hb, info, reply);
				break;
			}
			update_reset(CCM_GET_UPDATETABLE(info));
			update_add(CCM_GET_UPDATETABLE(info), CCM_GET_LLM(info),
						orig, uptime_val, TRUE);

			CCM_SET_MINORTRANS(info, trans_minorval);
			repeat = 0;
			while (ccm_send_join(hb, info) != HA_OK) {
				if(repeat < REPEAT_TIMES){
					ccm_debug(LOG_WARNING,
					"%s: failure to send join", __FUNCTION__);
					cl_shortsleep();
					repeat++;
				}else{
					break;
				}
			}
			ccm_set_state(info, CCM_STATE_JOINING, reply);
			break;		

		case CCM_TYPE_LEAVE:
			
			/* if the dead node is leader, jump to CCM state machine */
			if(node_is_leader(info, orig)){
				update_reset(CCM_GET_UPDATETABLE(info));
				CCM_INCREMENT_MINORTRANS(info);
				repeat = 0;
				while (ccm_send_join(hb, info) != HA_OK) {
					if(repeat < REPEAT_TIMES){
						ccm_debug(LOG_WARNING,
						"%s: failure to send join", __FUNCTION__);
						cl_shortsleep();
						repeat++;
					}else{
						break;
					}
				}
				ccm_set_state(info, CCM_STATE_JOINING, reply);
			}

		case CCM_TYPE_ALIVE:
			/* We do nothing here because we believe leader
			 * will deal with this LEAVE message. SPOF?
			 */
			break;		

		case CCM_TYPE_PROTOVERSION:
			/* we are waiting for the leader for membership list
			 * it's ok if someone want to join -- just ignore
			 * the message and let the leader handl it
			 */
			
			break;

		case CCM_TYPE_PROTOVERSION_RESP:
			break;
		case CCM_TYPE_STATE_INFO:
			ccm_handle_state_info(hb, info, reply);
			break;
		case CCM_TYPE_RESTART:
			ccm_all_restart(hb, info, reply);
			break;
		default:
			ccm_log(LOG_ERR,"%s: dropping message"
				" of type %s. Is this a Byzantine failure?", 
				__FUNCTION__, ccm_type2string(ccm_msg_type));
			/* nothing to do. Just forget the message */
			break;	
	}
}


static void ccm_fill_update_table(ccm_info_t *info,
		ccm_update_t *update_table, const void *uptime_list)
{
	const int *uptime;
	int i;

	uptime = (const int *)uptime_list;

	UPDATE_SET_NODECOUNT(update_table, info->memcount);
	for (i = 0; i< info->memcount; i++){
		update_table->update[i].index = info->ccm_member[i];
		update_table->update[i].uptime = ntohl(uptime[i]);
	}
	return;
} 

int
jump_to_joining_state(ll_cluster_t *hb, 
		      ccm_info_t *info,
		      struct ha_msg* msg){
	
	reset_change_info(info);
	update_reset(CCM_GET_UPDATETABLE(info));
	CCM_INCREMENT_MINORTRANS(info);
	if (ccm_send_join(hb, info) != HA_OK){
		ccm_log(LOG_ERR, "sending joining message failed");
		return HA_FAIL;
		
	}
	ccm_set_state(info, CCM_STATE_JOINING, msg);
	return HA_OK;
}

state_msg_handler_t	state_msg_handler[]={
	ccm_state_none,
	ccm_state_version_request,
	ccm_state_joining,  
	ccm_state_sent_memlistreq,
	ccm_state_memlist_res,
	ccm_state_joined, 
	ccm_state_wait_for_mem_list,
	ccm_state_wait_for_change,
	ccm_state_new_node_wait_for_mem_list,
};
	

static void
dump_mbrs(ccm_info_t *info)
{
	int i;
	const char *nodename;
	int leader;
	
	static struct born_s  {
		int index;
		int bornon;
	}  bornon[MAXNODE];/*avoid making it a 
				stack variable*/
	

	if(ccm_get_memcount(info)==1){
		bornon[0].index  = CCM_GET_MEMINDEX(info,0);
		bornon[0].bornon = CCM_GET_MAJORTRANS(info);
	} else for(i=0; i < ccm_get_memcount(info); i++){
		bornon[i].index = CCM_GET_MEMINDEX(info,i);
		bornon[i].bornon = update_get_uptime(CCM_GET_UPDATETABLE(info), 
						     CCM_GET_LLM(info),
						     CCM_GET_MEMINDEX(info,i));
		if(bornon[i].bornon==0) 
			bornon[i].bornon=CCM_GET_MAJORTRANS(info);
	}
	
	ccm_debug(LOG_DEBUG,"dump current membership %p", info);
	leader = info->ccm_cluster_leader;
	ccm_debug(LOG_DEBUG,"\tleader=%s"
	,	leader < 0 ?"none": info->llm.nodes[leader].nodename);
	ccm_debug(LOG_DEBUG,"\ttransition=%d", CCM_GET_MAJORTRANS(info));
	ccm_debug(LOG_DEBUG,"\tstatus=%s",state2string(info->state));
	ccm_debug(LOG_DEBUG,"\thas_quorum=%d",info->has_quorum);
	
	for (i=0 ;  i < ccm_get_memcount(info); i++) {
		nodename = llm_get_nodename(CCM_GET_LLM(info), 
					  CCM_GET_MEMINDEX(info,i));
		ccm_debug(LOG_DEBUG,"\tnodename=%s bornon=%d", nodename, 
		       bornon[i].bornon);
	}
	
	return;
}
void
ccm_on_quorum_changed(void)
{
	ccm_debug(LOG_DEBUG,"quorum changed");
	if (ccm_info_saved->state != CCM_STATE_JOINED) {
		ccm_debug(LOG_DEBUG,"we are not in CCM_STATE_JOINED, ignore");
		return;
	}
	send_mem_list_to_all(hb_fd_saved, ccm_info_saved, ccm_info_saved->ccm_cookie);
	report_mbrs(ccm_info_saved);
}