cl_msg.c

linux集群服务器软件代码包

/* Return the next message found in the IPC channel */
IPC_Message *
ipcmsgfromIPC(IPC_Channel * ch)
{
	int		rc;
	IPC_Message*	ipcmsg;

	rc = ch->ops->waitin(ch);

	switch(rc) {
		default:
		case IPC_FAIL:
			cl_perror("msgfromIPC: waitin failure");
			return NULL;

		case IPC_BROKEN:
			sleep(1);
			return NULL;

		case IPC_INTR:
			return NULL;

		case IPC_OK:
			break;
	}


	ipcmsg = NULL;
	rc = ch->ops->recv(ch, &ipcmsg);
#if 0
	if (DEBUGPKTCONT) {
		cl_log(LOG_DEBUG, "msgfromIPC: recv returns %d ipcmsg = 0x%lx"
		,	rc, (unsigned long)ipcmsg);
	}
#endif
	if (rc != IPC_OK) {
		return NULL;
	}


	return(ipcmsg);
}


/* Writes a message into a stream - used for serial lines */
int
msg2stream(struct ha_msg* m, FILE * f)
{
	size_t	len;
	char *	s  = msg2wirefmt(m, &len);

	if (s != NULL) {
		int	rc = HA_OK;
		if (fputs(s, f) == EOF) {
			rc = HA_FAIL;
			cl_perror("msg2stream: fputs failure");
		}
		if (fflush(f) == EOF) {
			cl_perror("msg2stream: fflush failure");
			rc = HA_FAIL;
		}
		ha_free(s);
		return(rc);
	}else{
		return(HA_FAIL);
	}
}
static void ipcmsg_done(IPC_Message* m);

static void
ipcmsg_done(IPC_Message* m)
{
	if (!m) {
		return;
	}
	if (m->msg_buf) {
		ha_free(m->msg_buf);
	}
	ha_free(m);
	m = NULL;
}


/*
 * create an ipcmsg and copy the data
 */

IPC_Message*
wirefmt2ipcmsg(void* p, size_t len, IPC_Channel* ch)
{
	IPC_Message*	ret = NULL;

	if (p == NULL){
		return(NULL);
	}

	ret = MALLOCT(IPC_Message);
	if (!ret) {
		return(NULL);
	}
	
	memset(ret, 0, sizeof(IPC_Message));
	
	if (NULL == (ret->msg_buf = cl_malloc(len + ch->msgpad))) {
		cl_free(ret);
		return NULL;
	}
	ret->msg_body = (char*)ret->msg_buf + ch->msgpad;
	memcpy(ret->msg_body, p, len);
	
	ret->msg_done = ipcmsg_done;
	ret->msg_private = NULL;
	ret->msg_ch = ch;
	ret->msg_len = len;

	return ret;

}

IPC_Message*
hamsg2ipcmsg(struct ha_msg* m, IPC_Channel* ch)
{
	size_t		len;
	char *		s  = msg2wirefmt(m, &len);
	IPC_Message*	ret = NULL;


	if (s == NULL) {
		return ret;
	}
	ret = MALLOCT(IPC_Message);
	if (!ret) {
		ha_free(s);
		return ret;
	}
	
	memset(ret, 0, sizeof(IPC_Message));

	if (NULL == (ret->msg_buf = cl_malloc(len + ch->msgpad))) {
		cl_free(s);
		cl_free(ret);
		return NULL;
	}
	ret->msg_body = (char*)ret->msg_buf + ch->msgpad;
	memcpy(ret->msg_body, s, len);
	cl_free(s);
	
	ret->msg_done = ipcmsg_done;
	ret->msg_private = NULL;
	ret->msg_ch = ch;
	ret->msg_len = len;

	return ret;
}

struct ha_msg*
ipcmsg2hamsg(IPC_Message*m)
{
	struct ha_msg*	ret = NULL;


	ret = wirefmt2msg(m->msg_body, m->msg_len);
	return ret;
}

int
msg2ipcchan(struct ha_msg*m, IPC_Channel*ch)
{
	IPC_Message*	imsg;

	if (m == NULL || ch == NULL) {
		cl_log(LOG_ERR, "Invalid msg2ipcchan argument");
		errno = EINVAL;
		return HA_FAIL;
	}

	if ((imsg = hamsg2ipcmsg(m, ch)) == NULL) {
		cl_log(LOG_ERR, "hamsg2ipcmsg() failure");
		return HA_FAIL;
	}

	if (ch->ops->send(ch, imsg) != IPC_OK) {
		if (ch->ch_status == IPC_CONNECT) {
			cl_log(LOG_ERR
			,	"msg2ipcchan: ch->ops->send() failure");
		}
		imsg->msg_done(imsg);
		return HA_FAIL;
	}
	return HA_OK;
}

static gboolean (*msg_authentication_method)(const struct ha_msg* ret) = NULL;


void
cl_set_oldmsgauthfunc(gboolean (*authfunc)(const struct ha_msg*))
{
	msg_authentication_method = authfunc;
}


/* Converts a string (perhaps received via UDP) into a message */
struct ha_msg *
string2msg_ll(const char * s, size_t length, int depth, int need_auth)
{
	struct ha_msg*	ret;
	int		startlen;
	int		endlen;
	const char *	sp = s;
	const char *	smax = s + length;


	if ((ret = ha_msg_new(0)) == NULL) {
		return(NULL);
	}

	startlen = sizeof(MSG_START)-1;
	if (strncmp(sp, MSG_START, startlen) != 0) {
		/* This can happen if the sender gets killed */
		/* at just the wrong time... */
		if (!cl_msg_quiet_fmterr) {
			cl_log(LOG_WARNING, "string2msg_ll: no MSG_START");
		}
		ha_msg_del(ret);
		return(NULL);
	}else{
		sp += startlen;
	}

	endlen = sizeof(MSG_END)-1;

	/* Add Name=value pairs until we reach MSG_END or end of string */

	while (*sp != EOS && strncmp(sp, MSG_END, endlen) != 0) {

		if (sp >= smax)	{
			return(NULL);
		}
		/* Skip over initial CR/NL things */
		sp += strspn(sp, CRNL);
		if (sp >= smax)	{
			return(NULL);
		}
		/* End of message marker? */
		if (strncmp(sp, MSG_END, endlen) == 0) {
			break;
		}
		/* Add the "name=value" string on this line to the message */
		if (ha_msg_add_nv_depth(ret, sp, smax, depth) != HA_OK) {
			if (!cl_msg_quiet_fmterr) {
				cl_log(LOG_ERR, "NV failure (string2msg_ll):");
				cl_log(LOG_ERR, "Input string: [%s]", s);
				cl_log(LOG_ERR, "sp=%s", sp);
			}
			ha_msg_del(ret);
			return(NULL);
		}
		if (sp >= smax) {
			return(NULL);
		}
		sp += strcspn(sp, CRNL);
	}

	if (need_auth && msg_authentication_method
	&&		!msg_authentication_method(ret)) {
		const char* from = ha_msg_value(ret, F_ORIG);
		if (!cl_msg_quiet_fmterr) {
			cl_log(LOG_WARNING
		       ,       "string2msg_ll: node [%s]"
		       " failed authentication", from ? from : "?");
		}
		ha_msg_del(ret);
		ret = NULL;
	}
	return(ret);
}



struct ha_msg *
string2msg(const char * s, size_t length)
{
	return(string2msg_ll(s, length, 0, NEEDAUTH));
}






/* Converts a message into a string (for sending out UDP interface)
   
   used in two places:

   1.called by msg2string as a implementation for computing string for a
   message provided the buffer
   
   2.called by is_authentic. In this case, there are no start/end string
   and the "auth" field is not included in the string

*/


int
msg2string_buf(const struct ha_msg *m, char* buf, size_t len
,	int depth,int needhead)
{

	char *	bp = NULL;
	int	j;
	char* maxp = buf + len;

	buf[0]=0;
	bp = buf;

	if (needhead){
		strcpy(bp, MSG_START);
		bp += strlen(MSG_START);
	}

	for (j=0; j < m->nfields; ++j) {
		
		int truelen;
		int (*tostring)(char*, char*, void*, size_t, int);	

		if (needhead == NOHEAD && strcmp(m->names[j], F_AUTH) == 0) {
			continue;
		}

		if (m->types[j] != FT_STRING){
			strcat(bp, "(");
			bp++;
			strcat(bp,FT_strings[m->types[j]]);
			bp++;
			strcat(bp,")");
			bp++;
		}

		strcat(bp, m->names[j]);
		bp += m->nlens[j];
		strcat(bp, "=");
		bp++;
		
		if(m->types[j] < DIMOF(fieldtypefuncs)){
			tostring = fieldtypefuncs[m->types[j]].tostring;
		} else {
			cl_log(LOG_ERR, "type(%d) unrecognized", m->types[j]);
			return HA_FAIL;
		}
		if (!tostring ||
		    (truelen = tostring(bp, maxp, m->values[j], m->vlens[j], depth))
		    < 0){
			cl_log(LOG_ERR, "tostring failed");
			return HA_FAIL;			
		}
		
		bp +=truelen;
		
		strcat(bp,"\n");
		bp++;


	}
	if (needhead){
		strcat(bp, MSG_END);
		bp += strlen(MSG_END);
	}

	bp[0] = 0;

	if (bp > buf + len){

		cl_log(LOG_ERR, "msg2string_buf: out of memory bound"
		", bp=%p, buf + len=%p, len=%ld"
		,	bp, buf + len, (long)len);

		cl_log_message(LOG_ERR, m);

		return(HA_FAIL);

	}

	return(HA_OK);
}


char *
msg2string(const struct ha_msg *m)
{
	void	*buf;
	int	len;

	AUDITMSG(m);
	if (m->nfields <= 0) {
		cl_log(LOG_ERR, "msg2string: Message with zero fields");
		return(NULL);
	}
	
	len = get_stringlen(m);
	
	if (len >= MAXMSG){
		cl_log(LOG_ERR, "msg2string: msg is too large"
		       "len =%d,MAX msg allowed=%d", len, MAXMSG);
		return NULL;
	}
	
	buf = ha_malloc(len);


	if (buf == NULL) {
		cl_log(LOG_ERR, "msg2string: no memory for string");
		return(NULL);
	}

	if (msg2string_buf(m, buf, len ,0, NEEDHEAD) != HA_OK){
		cl_log(LOG_ERR, "msg2string: msg2string_buf failed");
		ha_free(buf);
		return(NULL);
	}
	
	return(buf);
}


int
get_stringlen(const struct ha_msg *m)
{
	int i;
	int total_len =0 ;

	if (m == NULL){
		cl_log(LOG_ERR, "get_stringlen:"
		       "asking stringlen of a NULL message");
		return 0;
	}
	
	for (i = 0; i < m->nfields; i++){		
		if (m->types[i] == FT_STRUCT){			
			total_len += struct_stringlen(m->nlens[i], 
						      m->vlens[i],
						      m->values[i]);
		}
	}
	
	total_len += m->stringlen;
	
	return total_len;
}

int
get_netstringlen(const struct ha_msg *m)
{
	int i;
	int total_len =0 ;
	
	if (m == NULL){
		cl_log(LOG_ERR, "get_netstringlen:"
		       "asking netstringlen of a NULL message");
		return 0;
	}
	
	for (i = 0; i < m->nfields; i++){		
		if (m->types[i] == FT_STRUCT){			
			total_len += struct_netstringlen(m->nlens[i],
							 m->vlens[i],
							 m->values[i]);
		}
	}
	
	total_len += m->netstringlen;
	
	return total_len;	


}

char*
msg2wirefmt(const struct ha_msg*m, size_t* len)
{

	if (msgfmt == MSGFMT_NETSTRING) {
		return(msg2netstring(m, len));
	}
	else{
		char	*tmp;

		tmp = msg2string(m);
		
		if(tmp == NULL){
			*len = 0;
			return NULL;
		}
		
		*len = strlen(tmp) + 1;
		return(tmp);
	}
}


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

	int startlen;

	startlen = sizeof(MSG_START)-1;
	if (strncmp(s, MSG_START, startlen) == 0) {
		return(string2msg_ll(s, length, 0, need_auth));
	}

	startlen = sizeof(MSG_START_NETSTRING) - 1;
	if (strncmp(s, MSG_START_NETSTRING, startlen) == 0) {
		return netstring2msg(s, length, need_auth);
	}

	return NULL;

}




struct ha_msg*
wirefmt2msg(const char* s, size_t length)
{