ccmlib_memapi.c

linux集群服务器软件代码包

/* $Id: ccmlib_memapi.c,v 1.25 2005/02/17 19:08:20 gshi Exp $ */
/* 
 * ccmlib_memapi.c: Consensus Cluster Membership API
 *
 * Copyright (c) International Business Machines  Corp., 2002
 * Author: Ram Pai ([e-mail])
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * 
 * This software 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
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

#define __CCM_LIBRARY__
#include <ccmlib.h>
#include <ccm.h>
/*#include <syslog.h> */
#include <clplumbing/cl_log.h> 

/* structure to track the membership delivered to client */
typedef struct mbr_track_s {
	int			quorum;
	int			m_size;
	oc_ev_membership_t 	m_mem;
} mbr_track_t;


typedef struct mbr_private_s {
	int			magiccookie;
	gboolean		client_report; 	   /* report to client */
	oc_ev_callback_t 	*callback; /* the callback function registered
					      	by the client */
	struct IPC_CHANNEL      *channel; /* the channel to talk to ccm */
	ccm_llm_t 		*llm;	 /* list of all nodes */
	GHashTable  		*bornon; /* list of born time 
					    for all nodes */
	void 			*cookie; /* the last known 
					    membership event cookie */
	gboolean		special; /* publish non primary membership. 
					  * This is a kludge to accomodate 
					  * special behaviour not provided 
					  * but desired from the 0.2 API. 
					  * By default this behaviour is 
					  * turned off.
					  */
} mbr_private_t;


#define OC_EV_SET_INSTANCE(m,trans)  m->m_mem.m_instance=trans
#define OC_EV_SET_N_MEMBER(m,n)  m->m_mem.m_n_member=n
#define OC_EV_SET_MEMB_IDX(m,idx)  m->m_mem.m_memb_idx=idx
#define OC_EV_SET_N_OUT(m,n)  m->m_mem.m_n_out=n
#define OC_EV_SET_OUT_IDX(m,idx)  m->m_mem.m_out_idx=idx
#define OC_EV_SET_N_IN(m,n)  m->m_mem.m_n_in=n
#define OC_EV_SET_IN_IDX(m,idx)  m->m_mem.m_in_idx=idx
#define OC_EV_SET_NODEID(m,idx,nodeid)  m->m_mem.m_array[idx].node_id=nodeid
#define OC_EV_SET_BORN(m,idx,born)  m->m_mem.m_array[idx].node_born_on=born

#define OC_EV_INC_N_MEMBER(m)  m->m_mem.m_n_member++
#define OC_EV_INC_N_IN(m)  m->m_mem.m_n_in++
#define OC_EV_INC_N_OUT(m)  m->m_mem.m_n_out++


#define OC_EV_SET_SIZE(m,size)  m->m_size=size
#define OC_EV_SET_DONEFUNC(m,f)  m->m_func=f

#define OC_EV_GET_INSTANCE(m)  m->m_mem.m_instance
#define OC_EV_GET_N_MEMBER(m)  m->m_mem.m_n_member
#define OC_EV_GET_MEMB_IDX(m)  m->m_mem.m_memb_idx
#define OC_EV_GET_N_OUT(m)  m->m_mem.m_n_out
#define OC_EV_GET_OUT_IDX(m)  m->m_mem.m_out_idx
#define OC_EV_GET_N_IN(m)  m->m_mem.m_n_in
#define OC_EV_GET_IN(m)  m->m_mem.m_in_idx
#define OC_EV_GET_NODEARRY(m)  m->m_mem.m_array
#define OC_EV_GET_NODE(m,idx)  m->m_mem.m_array[idx]
#define OC_EV_GET_NODEID(m,idx)  m->m_mem.m_array[idx].node_id
#define OC_EV_GET_BORN(m,idx)  m->m_mem.m_array[idx].node_born_on

#define OC_EV_COPY_NODE_WITHOUT_UNAME(m1,idx1,m2,idx2)  \
		m1->m_mem.m_array[idx1]=m2->m_mem.m_array[idx2]

#define OC_EV_COPY_NODE(m1,idx1,m2,idx2)  \
		m1->m_mem.m_array[idx1]=m2->m_mem.m_array[idx2]; \
		m1->m_mem.m_array[idx1].node_uname = \
				strdup(m2->m_mem.m_array[idx2].node_uname)

#define OC_EV_GET_SIZE(m)  m->m_size


/* prototypes of external functions used in this file 
 * Should be made part of some header file 
 */
void *cookie_construct(void (*f)(void *), void (*free_f)(void *), void *);
void * cookie_get_data(void *ck);
void * cookie_get_func(void *ck);
void cookie_ref(void *ck);
void cookie_unref(void *ck);
static const char *llm_get_Id_from_Uuid(ccm_llm_t *stuff, uint uuid);



static void
init_llm(mbr_private_t *mem, struct IPC_MESSAGE *msg)
{
	unsigned long len = msg->msg_len;
	int	numnodes;

	mem->llm = (ccm_llm_t *)g_malloc(len);
	memcpy(mem->llm, msg->msg_body, len);

	mem->bornon = g_hash_table_new(g_direct_hash, 
				g_direct_equal);

	numnodes = CLLM_GET_NODECOUNT(mem->llm);
	mem->cookie = NULL;


	return;
}

static void
init_bornon(mbr_private_t *private, 
		struct IPC_MESSAGE *msg)
{
	ccm_born_t *born;
	int numnodes, i, n;
	struct born_s *bornon;

	numnodes = CLLM_GET_NODECOUNT(private->llm);

	born = (ccm_born_t *)msg->msg_body;

	n = born->n;
	assert(msg->msg_len == sizeof(ccm_born_t)
			+n*sizeof(struct born_s));
	bornon = born->born;
	for (i = 0 ; i < n; i++) {
		assert(bornon[i].index <= numnodes);
		g_hash_table_insert(private->bornon, 
			GINT_TO_POINTER(CLLM_GET_UUID(private->llm, 
					bornon[i].index)),
			GINT_TO_POINTER(bornon[i].bornon+1));
	}
	return;
}


static void
reset_bornon(mbr_private_t *private)
{
	g_hash_table_destroy(private->bornon);
	private->bornon = NULL;
}

static void
reset_llm(mbr_private_t *private)
{
	g_free(private->llm);
	private->llm = NULL;
}

static int
init_llmborn(mbr_private_t *private)
{
	fd_set rset;
	struct IPC_CHANNEL *ch;
	int 	sockfd, i=0, ret;
	struct IPC_MESSAGE *msg;
	struct timeval tv;

	if(private->llm) {
		return 0;
	}

	ch 	   = private->channel;
	sockfd = ch->ops->get_recv_select_fd(ch);

	/* receive the initiale low level membership 
	*  information in the first iteration, and 
	*  recieve the bornon information in the 
	*  second iteration 
	*/
	while( i < 2) {

		FD_ZERO(&rset);
		FD_SET(sockfd,&rset);
		tv.tv_sec = 1;
		tv.tv_usec = 0;

		if(!ch->ops->is_message_pending(ch) && (select(sockfd + 1, 
				&rset, NULL,NULL,&tv)) == -1){
			perror("select");
			ch->ops->destroy(ch);
			return -1;
		}
		ret = ch->ops->recv(ch,&msg);
		if(ret == IPC_BROKEN) {
			fprintf(stderr, "connection denied\n");
			return -1;
		}
	 	if(ret == IPC_FAIL){
			fprintf(stderr, ".");
			cl_shortsleep();
			continue;
		}

		switch(i) {
		case 0: init_llm(private, msg);
			break;
		case 1: init_bornon(private, msg);
			private->client_report = TRUE;
			break;
		}
		i++;
		msg->msg_done(msg);
	}
	return 0;
}


static gboolean
class_valid(class_t *class)
{
	mbr_private_t 	*private;

	if(class->type != OC_EV_MEMB_CLASS) {
		return FALSE;
	}

	private = (mbr_private_t *)class->private;

	if(!private || 
		private->magiccookie != 0xabcdef){
		return FALSE;
	}
	return TRUE;
}



static gboolean
already_present(oc_node_t *arr, uint size, oc_node_t node)
{
	uint i;
	for ( i = 0 ; i < size ; i ++ ) {
		if(arr[i].node_id == node.node_id &&
			arr[i].node_born_on >= node.node_born_on) {
			return TRUE;
		}
	}
	return FALSE;
}

static int
compare(const void *value1, const void *value2)
{
	const oc_node_t *t1 = (const oc_node_t *)value1;
	const oc_node_t *t2 = (const oc_node_t *)value2;
	
	if (t1->node_born_on < t2->node_born_on){
		return -1;
	}

	if (t1->node_born_on > t2->node_born_on){
		return 1;
	}

	if (t1->node_id < t2->node_id) {
		return -1;
	}

	if (t1->node_id > t2->node_id) {
		return 1;
	}
	
	return 0;
}

static const char *
llm_get_Id_from_Uuid(ccm_llm_t *stuff, uint uuid)
{
	uint lpc = 0;
	for (; lpc < stuff->n; lpc++) {
		if(stuff->node[lpc].Uuid == uuid){		       
			return stuff->node[lpc].Id;
		}
	}
	return NULL;
}



static int
get_new_membership(mbr_private_t *private,
		ccm_meminfo_t *mbrinfo,
		int		len,
		mbr_track_t **mbr)
{
	mbr_track_t *newmbr, *oldmbr;
	int trans, i, j, in_index, out_index, born;
	int n_members,uuid;
	
	int n_nodes = CLLM_GET_NODECOUNT(private->llm);

	int size    = sizeof(oc_ev_membership_t) + 
		2*n_nodes*sizeof(oc_node_t);

 	newmbr = *mbr = (mbr_track_t *) g_malloc(size+sizeof(int));

	trans = OC_EV_SET_INSTANCE(newmbr,mbrinfo->trans);
	n_members = OC_EV_SET_N_MEMBER(newmbr,mbrinfo->n);
	OC_EV_SET_SIZE(newmbr, size);

	j = OC_EV_SET_MEMB_IDX(newmbr,0);

	for ( i = 0 ; i < n_members; i++ ) {
		const char *uname = NULL;
		gpointer	gborn; /* Help make gcc warning go away */
		uuid =  CLLM_GET_UUID(private->llm, mbrinfo->member[i]);

		uname = llm_get_Id_from_Uuid(private->llm, uuid);

		newmbr->m_mem.m_array[j].node_uname = strdup(uname); 

		OC_EV_SET_NODEID(newmbr,j,uuid);

		gborn = g_hash_table_lookup(private->bornon, 
				GINT_TO_POINTER(mbrinfo->member[i]));
		born = GPOINTER_TO_INT(gborn);

		/* if there is already a born entry for the
		 * node, use it. Otherwise create a born entry
		 * for the node.
	 	 *
		 * NOTE: born==0 implies the entry has not been
		 * 	initialized.
		 */
		OC_EV_SET_BORN(newmbr,j, born==0?trans:(born-1));
		j++;
	}
	/* sort the m_arry */
	qsort(OC_EV_GET_NODEARRY(newmbr), n_members, 
			sizeof(oc_node_t), compare);

	in_index = OC_EV_SET_IN_IDX(newmbr,j);
	out_index = OC_EV_SET_OUT_IDX(newmbr,(j+n_nodes));

	OC_EV_SET_N_IN(newmbr,0);
	OC_EV_SET_N_OUT(newmbr,0);

	oldmbr = (mbr_track_t *) cookie_get_data(private->cookie);

	if(oldmbr) {
		for ( i = 0 ; i < n_members; i++ ) {
			if(!already_present(OC_EV_GET_NODEARRY(oldmbr),
					OC_EV_GET_N_MEMBER(oldmbr),
					OC_EV_GET_NODE(newmbr,i))){
				OC_EV_COPY_NODE_WITHOUT_UNAME(newmbr
				,	in_index, newmbr, i);
				in_index++;
				OC_EV_INC_N_IN(newmbr);
			}
		}

		for ( i = 0 ; (uint)i < OC_EV_GET_N_MEMBER(oldmbr) ; i++ ) {
			if(!already_present(OC_EV_GET_NODEARRY(newmbr), 
					OC_EV_GET_N_MEMBER(newmbr), 
					OC_EV_GET_NODE(oldmbr,i))){
				OC_EV_COPY_NODE(newmbr, out_index, oldmbr, i);
				out_index++;
				OC_EV_INC_N_OUT(newmbr);
			}
		}
	} else {
		OC_EV_SET_IN_IDX(newmbr,0);
		OC_EV_SET_N_IN(newmbr,OC_EV_GET_N_MEMBER(newmbr));
	}

	return size;
}

static void
mem_free_func(void *data)
{
	unsigned lpc = 0;
	char * uname;
	mbr_track_t  *mbr_track =  (mbr_track_t *)data;

	if(mbr_track) {
		/* free m_n_member uname, m_n_in is actually the same ptr */
		for (lpc = 0 ; lpc < OC_EV_GET_N_MEMBER(mbr_track); lpc++ ) {
			if ((uname = OC_EV_GET_NODE(mbr_track, lpc).node_uname)){
				g_free(uname);
			}
		}
		/* free m_n_out uname */
		for (lpc = OC_EV_GET_OUT_IDX(mbr_track)
		;	lpc < OC_EV_GET_OUT_IDX(mbr_track)
			+	OC_EV_GET_N_OUT(mbr_track)
		;	lpc++) {