cl_msg.c

在LINUX下实现HA的源代码

/* $Id: cl_msg.c,v 1.6.2.12 2005/01/03 20:28:45 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>

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


#define		NL_TO_SYM	0
#define		SYM_TO_NL	1
#define		NEEDAUTH	1
#define		NOAUTH		0

static enum cl_msgfmt msgfmt = MSGFMT_NVPAIR;

int		SPECIAL_SYMS[]={
	20,
	21,
	22,
	23,
	24,
	25,
	26,
	27,
	28,
	29
};


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);
static struct ha_msg* string2msg_ll(const char * s, size_t length, int need_auth, int depth);

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;
}

/*
 * This function changes each new line in the input string
 * into a special symbol, or the other way around
 */

static int
convert(char* s, int len, int depth, int direction)
{
	int	i;

	if (direction != NL_TO_SYM && direction != SYM_TO_NL){
		cl_log(LOG_ERR, "convert(): direction not defined!");
		return(HA_FAIL);
	}


	if (depth >= MAXDEPTH ){
		cl_log(LOG_ERR, "convert(): MAXDEPTH exceeded");
		return(HA_FAIL);
	}

	for (i = 0; i < len; i++){

		switch(direction){
		case NL_TO_SYM :
			if (s[i] == '\n'){
				s[i] = SPECIAL_SYMS[depth];
				break;
			}

			if (s[i] == SPECIAL_SYMS[depth]){
				cl_log(LOG_ERR
				, "convert(): special symbol found in string");
				return(HA_FAIL);
			}

			break;

		case SYM_TO_NL:

			if (s[i] == '\n'){
				cl_log(LOG_ERR
				,	"convert(): new line found in"
				" converted string");
				return(HA_FAIL);
			}

			if (s[i] == SPECIAL_SYMS[depth]){
				s[i] = '\n';
				break;
			}
			break;
		default:
			/* nothing, never executed*/;

		}
	}

	return(HA_OK);
}


static int
intlen(int x)
{
	char	buf[20];
	return snprintf(buf, sizeof(buf), "%d", x);
}


/* 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");
			ha_msg_del(ret);
			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] != FT_STRUCT) {
					ha_free(msg->values[j]);
				}else{
					ha_msg_del((struct ha_msg*)msg->values[j]);
				}
				msg->values[j] = NULL;
			}
			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 = MALLOCT(struct ha_msg)) == NULL) {
		return NULL;
	}
	ret->nfields	= msg->nfields;
	ret->nalloc	= msg->nalloc;
	ret->stringlen	= msg->stringlen;
	ret->netstringlen = msg->netstringlen;

	ret->names  = (char **)	ha_calloc(sizeof(char *), msg->nalloc);
	ret->nlens  = (int *)	ha_calloc(sizeof(int), msg->nalloc);
	ret->values = (void **)	ha_calloc(sizeof(void *), msg->nalloc);
	ret->vlens  = (size_t *)	ha_calloc(sizeof(size_t), msg->nalloc);
	ret->types  = (int *) ha_calloc(sizeof(int), msg->nalloc);
	if (ret->names == NULL || ret->values == NULL
	||	ret->nlens == NULL || ret->vlens == NULL || ret->types == NULL) {
		cl_log(LOG_ERR
		,	"ha_msg_new: out of memory for ha_msg_copy");
		goto freeandleave;
	}
	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 (ret->types[j] == FT_STRUCT){
			if ((ret->values[j]
			=	(void*)ha_msg_copy((struct ha_msg*)msg->values[j]))
			==	NULL){

				cl_log(LOG_ERR
				, "ha_msg_copy(): copy child message failed");
				goto freeandleave;
			}
		}else if ((ret->values[j] = ha_malloc(msg->vlens[j]+1))==NULL){
			goto freeandleave;
		}else{
			memcpy(ret->values[j], msg->values[j], msg->vlens[j]+1);
		}

	}
	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) {
		cl_log(LOG_CRIT
		,	"Message @ 0x%x has negative stringlen (%d)"
		,	(unsigned) msg, msg->stringlen);
		doabort = TRUE;
	}
	if (msg->stringlen < 4 * msg->nfields) {
		cl_log(LOG_CRIT
		,	"Message @ 0x%x has too small stringlen (%d)"
		,	(unsigned) msg, msg->stringlen);
		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

/* low level implementation for ha_msg_add

   the caller is responsible to allocate/free memories
   for @name and @value.

   @type could be FT_STRING, FT_BINARY, FT_STRUCT

   1. FT_STRING:
	In this case, this function could be called by heartbeat
   or by a client. If @name is a normal string, then @value is a normal
   string. Otherwise if @name is in format of (t), where type is an
   interget, then @value is base64 version of binary data (t == FT_BINARY)
   or converted string for a child message (t == FT_STRUCT).


   2. FT_BINARY
	@type equals to FT_BINARY implies it is called by a client
   to add a binary field to a message because heartbeat itself does not
   add any binary field.

   3. FT_STRUCT
	@type equals to FT_STRUCT implies it is called by a client
   to add a child message to this message because heartbeat itself does not
   add any child message.

*/

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

	int	next;
	size_t	startlen = sizeof(MSG_START)-1;
	size_t	startlen_netstring = sizeof(MSG_START_NETSTRING) -1 ;
	size_t	newstringlen;

	char	*cp_name = NULL;
	size_t	cp_namelen;
	size_t	cp_vallen;
	void	*cp_value = NULL;
	int	internal_type;

	if (!msg ||msg->names == NULL || msg->values == NULL) {
		cl_log(LOG_ERR
		,	"ha_msg_addraw_ll: cannot add field to ha_msg");
		return(HA_FAIL);
	}


        if (msg->nfields >= msg->nalloc) { 
                   char ** names = msg->names; 
                   int  *  nlens = msg->nlens; 
                   void ** values = msg->values; 
                   size_t* vlens = msg->vlens; 
                   int *   types = msg->types; 
    
                   int 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" 
                           ,       "ha_msg_addraw_ll: out of memory for ha_msg"); 
                           ha_msg_del(msg); 
    
                           cl_log(LOG_ERR, 
                                   "ha_msg_addraw_ll: cannot add field to 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; 
           } 



	if (namelen >= startlen && strncmp(name, MSG_START, startlen) == 0) {
		if (!cl_msg_quiet_fmterr) {
			cl_log(LOG_ERR, "ha_msg_addraw_ll: illegal field");
		}
		return(HA_FAIL);
	}
	if (namelen >= startlen_netstring
	&&	strncmp(name, MSG_START_NETSTRING, startlen_netstring) == 0){
		if (!cl_msg_quiet_fmterr) {
			cl_log(LOG_ERR, "ha_msg_addraw_ll: illegal field");
		}
	}

	if (name == NULL || value == NULL
	    ||	namelen <= 0 || vallen < 0) {
		cl_log(LOG_ERR, "ha_msg_addraw_ll: "
		       "cannot add name/value to ha_msg");
		return(HA_FAIL);
	}

	switch(type){

	case FT_BINARY:

		/* 3 == "(type)" and 2 == "=" + "\n" */
		newstringlen = msg->stringlen + (namelen + 3 + B64_stringlen(vallen)+2);

		cp_name = name;
		cp_namelen = namelen;
		cp_value = value;
		cp_vallen = vallen;
		internal_type = type;

		break;
	case FT_STRUCT:

		/* 3 == "(type)" and 2 == "=" + "\n" */
		newstringlen = msg->stringlen + namelen + 3 +  2;

		next = msg->nfields;
		msg->names[next] = name;
		msg->nlens[next] = namelen;
		msg->values[next] = value;
		msg->vlens[next] = vallen;
		msg->stringlen = newstringlen;
		msg->netstringlen += 0;
		/*  intlen(namelen) + (namelen) + intlen(vallen) + vallen + 4 */