cl_msg.c

linux集群服务器软件代码包

/* $Id: cl_msg.c,v 1.55 2005/02/17 21:45:35 gshi Exp $ */
/*
 * Heartbeat messaging object.
 *
 * Copyright (C) 2000 Alan Robertson <[e-mail]>
 *
 * This software licensed under the GNU LGPL.
 *
 * 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 library 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
 *
 */

#include <portability.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <errno.h>
#include <sys/utsname.h>
#include <ha_msg.h>
#include <unistd.h>
#include <clplumbing/cl_malloc.h>
#include <clplumbing/cl_log.h>
#include <clplumbing/ipc.h>
#include <clplumbing/base64.h>
#include <clplumbing/netstring.h>
#include <glib.h>
#include <uuid/uuid.h>

#define		MAXMSGLINE	MAXMSG
#define		MINFIELDS	30
#define		CRNL		"\r\n"
#define		MAX_AUTH_BYTES	64


#define		NEEDAUTH	1
#define		NOAUTH		0
#define		MAX_INT_LEN 	10
#define		MAX_NAME_LEN 	255
#define		UUID_SLEN	64
static enum cl_msgfmt msgfmt = MSGFMT_NVPAIR;



const char*
FT_strings[]={
	"0",
	"1",
	"2",
	"3",
	"4",
	"5",
	"6",
	"7",
	"8",
	"9"
};




#undef DOAUDITS
#ifdef DOAUDITS

void ha_msg_audit(const struct ha_msg* msg);

#	define	AUDITMSG(msg)	ha_msg_audit(msg)
#else
#	define	AUDITMSG(msg)	/* Nothing */
#endif


static volatile hb_msg_stats_t*	msgstats = NULL;

gboolean cl_msg_quiet_fmterr = FALSE;

extern int		netstring_format;

static struct ha_msg* wirefmt2msg_ll(const char* s, size_t length, int need_auth);

struct ha_msg* string2msg_ll(const char * s, size_t length, int need_auth, int depth);

extern int struct_stringlen(size_t namlen, size_t vallen, const void* value);
extern int struct_netstringlen(size_t namlen, size_t vallen, const void* value);


void
cl_msg_setstats(volatile hb_msg_stats_t* stats)
{
	msgstats = stats;
}

/* Set default messaging format */
void
cl_set_msg_format(enum cl_msgfmt mfmt)
{
	msgfmt = mfmt;
}



void
list_cleanup(GList* list)
{
	int i;
	for (i = 0; i < g_list_length(list); i++){
		char* element = g_list_nth_data(list, i);
		if (element == NULL){
			cl_log(LOG_WARNING, "list_cleanup:"
			       "element is NULL");
			continue;
		}
		ha_free(element);
	}
	g_list_free(list);
}



/* Create a new (empty) message */
struct ha_msg *
ha_msg_new(nfields)
{
	struct ha_msg *	ret;
	int	nalloc;

	ret = MALLOCT(struct ha_msg);
	if (ret) {
		ret->nfields = 0;

		if (nfields > MINFIELDS) {
			nalloc = nfields;
		} else {
			nalloc = MINFIELDS;
		}

		ret->nalloc    = nalloc;
		ret->names     = (char **)ha_calloc(sizeof(char *), nalloc);
		ret->nlens     = (int *)ha_calloc(sizeof(int), nalloc);
		ret->values    = (void **)ha_calloc(sizeof(void *), nalloc);
		ret->vlens     = (size_t *)ha_calloc(sizeof(size_t), nalloc);
		ret->stringlen = sizeof(MSG_START)+sizeof(MSG_END)-1;
		ret->netstringlen = sizeof(MSG_START_NETSTRING)
		+	sizeof(MSG_END_NETSTRING) - 1 + MAX_AUTH_BYTES;
		ret->types	= (int*)ha_calloc(sizeof(int), nalloc);

		if (ret->names == NULL || ret->values == NULL
		||	ret->nlens == NULL || ret->vlens == NULL
		||	ret->types == NULL) {

			cl_log(LOG_ERR, "%s"
			,	"ha_msg_new: out of memory for ha_msg");
			/* It is safe to give this to ha_msg_del() */
			/* at this point.  It's well-enough-formed */
			ha_msg_del(ret); /*violated property*/
			ret = NULL;
		}else if (msgstats) {
			msgstats->allocmsgs++;
			msgstats->totalmsgs++;
			msgstats->lastmsg = time_longclock();
		}
	}
	return(ret);
}

/* Delete (destroy) a message */
void
ha_msg_del(struct ha_msg *msg)
{
	if (msg) {
		int	j;
		AUDITMSG(msg);
		if (msgstats) {
			msgstats->allocmsgs--;
		}
		if (msg->names) {
			for (j=0; j < msg->nfields; ++j) {
				if (msg->names[j]) {
					ha_free(msg->names[j]);
					msg->names[j] = NULL;
				}
			}
			ha_free(msg->names);
			msg->names = NULL;
		}
		if (msg->values) {
			for (j=0; j < msg->nfields; ++j) {

				if (msg->values[j] == NULL){
					continue;
				}
				
				if(msg->types[j] < DIMOF(fieldtypefuncs)){					
					fieldtypefuncs[msg->types[j]].memfree(msg->values[j]);
				}
			}
			ha_free(msg->values);
			msg->values = NULL;
		}
		if (msg->nlens) {
			ha_free(msg->nlens);
			msg->nlens = NULL;
		}
		if (msg->vlens) {
			ha_free(msg->vlens);
			msg->vlens = NULL;
		}
		if (msg->types){
			ha_free(msg->types);
			msg->types = NULL;
		}
		msg->nfields = -1;
		msg->nalloc = -1;
		msg->stringlen = -1;
		msg->netstringlen = -1;
		ha_free(msg);
	}
}
struct ha_msg*
ha_msg_copy(const struct ha_msg *msg)
{
	struct ha_msg*		ret;
	int			j;

	
	AUDITMSG(msg);
	if (msg == NULL || (ret = ha_msg_new(msg->nalloc)) == NULL) {   
		return NULL;   
	} 

	ret->nfields	= msg->nfields;
	ret->stringlen	= msg->stringlen;
	ret->netstringlen = msg->netstringlen;

	memcpy(ret->nlens, msg->nlens, sizeof(msg->nlens[0])*msg->nfields);
	memcpy(ret->vlens, msg->vlens, sizeof(msg->nlens[0])*msg->nfields);
	memcpy(ret->types, msg->types, sizeof(msg->types[0])*msg->nfields);

	for (j=0; j < msg->nfields; ++j) {

		if ((ret->names[j] = ha_malloc(msg->nlens[j]+1)) == NULL) {
			goto freeandleave;
		}
		memcpy(ret->names[j], msg->names[j], msg->nlens[j]+1);
		
		
		if(msg->types[j] < DIMOF(fieldtypefuncs)){					
			ret->values[j] = fieldtypefuncs[msg->types[j]].dup(msg->values[j],
									   msg->vlens[j]);
			if (!ret->values[j]){
				cl_log(LOG_ERR,"duplicating the message field failed");
				goto freeandleave;
			}
		}
	}
	return ret;

freeandleave:
        /*   
	 * ha_msg_del nicely handles partially constructed ha_msgs
	 * so, there's not really a memory leak here at all, but BEAM   
	 * thinks there is.   
	 */   
	ha_msg_del(ret);/* memory leak */       ret=NULL; 
	return ret;
}

#ifdef DOAUDITS
void
ha_msg_audit(const struct ha_msg* msg)
{
	int	doabort = FALSE;
	int	j;

	if (!msg) {
		return;
	}
	if (!ha_is_allocated(msg)) {
		cl_log(LOG_CRIT, "Message @ 0x%x is not allocated"
		,	(unsigned) msg);
		abort();
	}
	if (msg->nfields < 0) {
		cl_log(LOG_CRIT, "Message @ 0x%x has negative fields (%d)"
		,	(unsigned) msg, msg->nfields);
		doabort = TRUE;
	}
	if (msg->nalloc < 0) {
		cl_log(LOG_CRIT, "Message @ 0x%x has negative nalloc (%d)"
		,	(unsigned) msg, msg->nalloc);
		doabort = TRUE;
	}
	if (msg->stringlen <=0 
	    || msg->netstringlen <= 0) {
		cl_log(LOG_CRIT,
		       "Message @ 0x%x has non-negative net/stringlen field"
		       "stringlen=(%d), netstringlen=(%d)",
		       (unsigned) msg, msg->stringlen, msg->netstringlen);
		doabort = TRUE;
	}

	if (!ha_is_allocated(msg->names)) {
		cl_log(LOG_CRIT
		,	"Message names @ 0x%x is not allocated"
		,	(unsigned) msg->names);
		doabort = TRUE;
	}
	if (!ha_is_allocated(msg->values)) {
		cl_log(LOG_CRIT
		,	"Message values @ 0x%x is not allocated"
		,	(unsigned) msg->values);
		doabort = TRUE;
	}
	if (!ha_is_allocated(msg->nlens)) {
		cl_log(LOG_CRIT
		,	"Message nlens @ 0x%x is not allocated"
		,	(unsigned) msg->nlens);
		doabort = TRUE;
	}
	if (!ha_is_allocated(msg->vlens)) {
		cl_log(LOG_CRIT
		,	"Message vlens @ 0x%x is not allocated"
		,	(unsigned) msg->vlens);
		doabort = TRUE;
	}
	if (doabort) {
		abort();
	}
	for (j=0; j < msg->nfields; ++j) {
		if (!ha_is_allocated(msg->names[j])) {
			cl_log(LOG_CRIT, "Message name[%d] @ 0x%x"
			" is not allocated."
			,	j, (unsigned) msg->names[j]);
		}
		if (!ha_is_allocated(msg->values[j])) {
			cl_log(LOG_CRIT, "Message value [%d] @ 0x%x"
			" is not allocated."
			,	j, (unsigned) msg->values[j]);
		}
	}
}
#endif



static int
ha_msg_expand(struct ha_msg* msg )
{	
	char **	names ;
	int  *	nlens ;
	void **	values ;
	size_t*	vlens ;
	int *	types ;
	int	nalloc;
       
	if(!msg){
		cl_log(LOG_ERR, "ha_msg_expand:"
		       "input msg is null");
		return HA_FAIL;
	}

	names = msg->names;
	nlens = msg->nlens;
	values = msg->values;
	vlens = msg->vlens;
	types = msg->types;
	
	nalloc = msg->nalloc + MINFIELDS;
	msg->names = 	(char **)ha_calloc(sizeof(char *), nalloc);
	msg->nlens = 	(int *)ha_calloc(sizeof(int), nalloc);
	msg->values = 	(void **)ha_calloc(sizeof(void *), nalloc);
	msg->vlens = 	(size_t *)ha_calloc(sizeof(size_t), nalloc);
	msg->types= 	(int*)ha_calloc(sizeof(int), nalloc);
	
	if (msg->names == NULL || msg->values == NULL
	    ||	msg->nlens == NULL || msg->vlens == NULL
	    ||	msg->types == NULL) {
		
		cl_log(LOG_ERR, "%s"
		       ,	" out of memory for ha_msg");		
		return(HA_FAIL);
	}
	
	memcpy(msg->names, names, msg->nalloc*sizeof(char *));
	memcpy(msg->nlens, nlens, msg->nalloc*sizeof(int));
	memcpy(msg->values, values, msg->nalloc*sizeof(void *));
	memcpy(msg->vlens, vlens, msg->nalloc*sizeof(size_t));
	memcpy(msg->types, types, msg->nalloc*sizeof(int));
	
	ha_free(names);
	ha_free(nlens);
	ha_free(values);
	ha_free(vlens);
	ha_free(types);
	
	msg->nalloc = nalloc;
	
	return HA_OK;
}

int
cl_msg_remove_value(struct ha_msg* msg, const void* value)
{
	int j;
	
	if (msg == NULL || value == NULL){
		cl_log(LOG_ERR, "cl_msg_remove: invalid argument");
		return HA_FAIL;
	}
	
	for (j = 0; j < msg->nfields; ++j){
		if (value == msg->values[j]){
			break;			
		}
	}
	if (j == msg->nfields){		
		cl_log(LOG_ERR, "cl_msg_remove: field %p not found", value);
		return HA_FAIL;
	}
	return cl_msg_remove_offset(msg, j);
	
}


int
cl_msg_remove(struct ha_msg* msg, const char* name)
{
	int j;
	
	if (msg == NULL || name == NULL){
		cl_log(LOG_ERR, "cl_msg_remove: invalid argument");
		return HA_FAIL;
	}
	
	for (j = 0; j < msg->nfields; ++j){
		if (strcmp(name, msg->names[j]) == 0){
			break;			
		}
	}
	
	if (j == msg->nfields){		
		cl_log(LOG_ERR, "cl_msg_remove: field %s not found", name);
		return HA_FAIL;
	}
	return cl_msg_remove_offset(msg, j);
}

int
cl_msg_remove_offset(struct ha_msg* msg, int offset)
{
	int j = offset;
	int i;
	int tmplen;
	
	if (j == msg->nfields){		
		cl_log(LOG_ERR, "cl_msg_remove: field %d not found", j);
		return HA_FAIL;
	}
	

	
	if (msg->types[j] != FT_STRUCT){
		tmplen = msg->stringlen;
		msg->stringlen -=  fieldtypefuncs[msg->types[j]].stringlen(msg->nlens[j],
									   msg->vlens[j],
									   msg->values[j]);
		if (msg->stringlen <=0){
			cl_log(LOG_ERR, "cl_msg_remove: stringlen <= 0 after removing"
			       "field %s. Return failure", msg->names[j]);
			msg->stringlen =tmplen;
			return HA_FAIL;
		}
		
		tmplen = msg->netstringlen;
		msg->netstringlen -=  fieldtypefuncs[msg->types[j]].netstringlen(msg->nlens[j],
										 msg->vlens[j],       
										 msg->values[j]);	
		if (msg->netstringlen <=0){
			cl_log(LOG_ERR, "cl_msg_remove: netstringlen <= 0 after removing"
			       "field %s. return failure", msg->names[j]);
			msg->netstringlen =tmplen;
			return HA_FAIL;
		}
	}

	
	ha_free(msg->names[j]);
	fieldtypefuncs[msg->types[j]].memfree(msg->values[j]);
	
	for (i= j + 1; i < msg->nfields ; i++){
		msg->names[i -1] = msg->names[i];
		msg->nlens[i -1] = msg->nlens[i];
		msg->values[i -1] = msg->values[i];
		msg->vlens[i-1] = msg->vlens[i];
		msg->types[i-1] = msg->types[i];
	}
	msg->nfields--;

	
	return HA_OK;
}



/* low level implementation for ha_msg_add
   the caller is responsible to allocate/free memories
   for @name and @value.

*/

static int
ha_msg_addraw_ll(struct ha_msg * msg, char * name, size_t namelen,
		 void * value, size_t vallen, int type, int depth)
{