mcast.c

linux集群服务器软件代码包

/* $Id: mcast.c,v 1.22 2004/10/24 13:00:13 lge Exp $ */
/*
 * mcast.c: implements hearbeat API for UDP multicast communication
 *
 * Copyright (C) 2000 Alan Robertson <[e-mail]>
 * Copyright (C) 2000 Chris Wright <[e-mail]>
 *
 * Thanks to WireX for providing hardware to test on.
 *
 * 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 <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <ctype.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <net/if.h>
#include <sys/ioctl.h>

#ifdef HAVE_SYS_SOCKIO_H
#	include <sys/sockio.h>
#endif

#include <HBcomm.h>
 
#define PIL_PLUGINTYPE          HB_COMM_TYPE
#define PIL_PLUGINTYPE_S        HB_COMM_TYPE_S
#define PIL_PLUGIN              mcast
#define PIL_PLUGIN_S            "mcast"
#define PIL_PLUGINLICENSE	LICENSE_LGPL
#define PIL_PLUGINLICENSEURL	URL_LGPL
#include <pils/plugin.h>
#include <heartbeat.h>

struct mcast_private {
	char *  interface;      /* Interface name */
	struct  in_addr mcast;  /* multicast address */
	struct  sockaddr_in   addr;   /* multicast addr */
	u_short port;
	int     rsocket;        /* Read-socket */
	int     wsocket;        /* Write-socket */
	u_char	ttl;		/* TTL value for outbound packets */
	u_char	loop;		/* boolean, loop back outbound packets */
};


static int		mcast_parse(const char* configline);
static struct hb_media * mcast_new(const char * intf, const char *mcast
			,	u_short port, u_char ttl, u_char loop);
static int		mcast_open(struct hb_media* mp);
static int		mcast_close(struct hb_media* mp);
static void*		mcast_read(struct hb_media* mp, int* lenp);
static int		mcast_write(struct hb_media* mp, void* p, int len);
static int		mcast_descr(char** buffer);
static int		mcast_mtype(char** buffer);
static int		mcast_isping(void);


static struct hb_media_fns mcastOps ={
	NULL,		/* Create single object function */
	mcast_parse,	/* whole-line parse function */
	mcast_open,
	mcast_close,
	mcast_read,
	mcast_write,
	mcast_mtype,
	mcast_descr,
	mcast_isping,
};

PIL_PLUGIN_BOILERPLATE2("1.0", Debug)
static const PILPluginImports*  PluginImports;
static PILPlugin*               OurPlugin;
static PILInterface*		OurInterface;
static struct hb_media_imports*	OurImports;
static void*			interfprivate;

#define LOG	PluginImports->log
#define MALLOC	PluginImports->alloc
#define STRDUP  PluginImports->mstrdup
#define FREE	PluginImports->mfree

PIL_rc
PIL_PLUGIN_INIT(PILPlugin*us, const PILPluginImports* imports);

PIL_rc
PIL_PLUGIN_INIT(PILPlugin*us, const PILPluginImports* imports)
{
	/* Force the compiler to do a little type checking */
	(void)(PILPluginInitFun)PIL_PLUGIN_INIT;

	PluginImports = imports;
	OurPlugin = us;

	/* Register ourself as a plugin */
	imports->register_plugin(us, &OurPIExports);  

	/*  Register our interface implementation */
 	return imports->register_interface(us, PIL_PLUGINTYPE_S
	,	PIL_PLUGIN_S
	,	&mcastOps
	,	NULL		/*close */
	,	&OurInterface
	,	(void*)&OurImports
	,	interfprivate); 
}


/* helper functions */
static int mcast_make_receive_sock(struct hb_media* hbm);
static int mcast_make_send_sock(struct hb_media * hbm);
static struct mcast_private *
new_mcast_private(const char *ifn, const char *mcast, u_short port,
		u_char ttl, u_char loop);
static int set_mcast_if(int sockfd, char *ifname);
static int set_mcast_loop(int sockfd, u_char loop);
static int set_mcast_ttl(int sockfd, u_char ttl);
static int join_mcast_group(int sockfd, struct in_addr *addr, char *ifname);
static int if_getaddr(const char *ifname, struct in_addr *addr);
static int is_valid_dev(const char *dev);
static int is_valid_mcast_addr(const char *addr);
static int get_port(const char *port, u_short *p);
static int get_ttl(const char *ttl, u_char *t);
static int get_loop(const char *loop, u_char *l);


#define		ISMCASTOBJECT(mp) ((mp) && ((mp)->vf == (void*)&mcastOps))
#define		MCASTASSERT(mp)	g_assert(ISMCASTOBJECT(mp))

static int
mcast_mtype(char** buffer)
{ 
	*buffer = STRDUP(PIL_PLUGIN_S);
	if (!*buffer) {
		return 0;
	}

	return STRLEN_CONST(PIL_PLUGIN_S);
}

static int
mcast_descr(char **buffer)
{ 
	const char cret[] = "UDP/IP multicast";
	*buffer = STRDUP(cret);
	if (!*buffer) {
		return 0;
	}

	return STRLEN_CONST(cret);
}

static int
mcast_isping(void)
{
	/* nope, this is not a ping device */
	return 0;
}

/* mcast_parse will parse the line in the config file that is 
 * associated with the media's type (hb_dev_mtype).  It should 
 * receive the rest of the line after the mtype.  And it needs
 * to call hb_dev_new, add the media to the list of available media.
 *
 * So in this case, the config file line should look like
 * mcast [device] [mcast group] [port] [mcast ttl] [mcast loop]
 * for example:
 * mcast eth0 225.0.0.1 694 1 0
 */

static int
mcast_parse(const char *line)
{
	const char *		bp = line;
	char			dev[MAXLINE];
	char			mcast[MAXLINE];
	char			token[MAXLINE];
	u_short			port = 0;	/* Bogus */
	u_char			ttl = 10;	/* Bogus */
	u_char			loop = 10;	/* Bogus */
	int			toklen;
	struct hb_media *	mp;

	/* Skip over white space, then grab the device */
	bp += strspn(bp, WHITESPACE);
	toklen = strcspn(bp, WHITESPACE);
	strncpy(dev, bp, toklen);
	bp += toklen;
	dev[toklen] = EOS;

	if (*dev != EOS)  {
		if (!is_valid_dev(dev)) {
			PILCallLog(LOG, PIL_CRIT, "mcast bad device [%s]", dev);
			return HA_FAIL;
		}
		/* Skip over white space, then grab the multicast group */
		bp += strspn(bp, WHITESPACE);
		toklen = strcspn(bp, WHITESPACE);
		strncpy(mcast, bp, toklen);
		bp += toklen;
		mcast[toklen] = EOS;
	
		if (*mcast == EOS)  {
			PILCallLog(LOG, PIL_CRIT, "mcast [%s] missing mcast address",
				dev);
			return(HA_FAIL);
		}
		if (!is_valid_mcast_addr(mcast)) {
			PILCallLog(LOG, PIL_CRIT, "mcast [%s] bad addr [%s]", dev, mcast);
			return(HA_FAIL);
		}

		/* Skip over white space, then grab the port */
		bp += strspn(bp, WHITESPACE);
		toklen = strcspn(bp, WHITESPACE);
		strncpy(token, bp, toklen);
		bp += toklen;
		token[toklen] = EOS;

		if (*token == EOS)  {
			PILCallLog(LOG, PIL_CRIT, "mcast [%s] missing port"
			,	dev);
			return(HA_FAIL);
		}
		if (get_port(token, &port) < 0 || port <= 0) {
			PILCallLog(LOG, PIL_CRIT, " mcast [%s] bad port [%d]", dev, port);
			return HA_FAIL;
		}

		/* Skip over white space, then grab the ttl */
		bp += strspn(bp, WHITESPACE);
		toklen = strcspn(bp, WHITESPACE);
		strncpy(token, bp, toklen);
		bp += toklen;
		token[toklen] = EOS;

		if (*token == EOS)  {
			PILCallLog(LOG, PIL_CRIT, "mcast [%s] missing ttl", dev);
			return(HA_FAIL);
		}
		if (get_ttl(token, &ttl) < 0 || ttl > 4) {
			PILCallLog(LOG, PIL_CRIT, " mcast [%s] bad ttl [%d]", dev, ttl);
			return HA_FAIL;
		}

		/* Skip over white space, then grab the loop */
		bp += strspn(bp, WHITESPACE);
		toklen = strcspn(bp, WHITESPACE);
		strncpy(token, bp, toklen);
		bp += toklen;
		token[toklen] = EOS;

		if (*token == EOS)  {
			PILCallLog(LOG, PIL_CRIT, "mcast [%s] missing loop", dev);
			return(HA_FAIL);
		}
		if (get_loop(token, &loop) < 0 ||	loop > 1) {
			PILCallLog(LOG, PIL_CRIT, " mcast [%s] bad loop [%d]", dev, loop);
			return HA_FAIL;
		}

		if ((mp = mcast_new(dev, mcast, port, ttl, loop)) == NULL) {
			return(HA_FAIL);
		}
		OurImports->RegisterNewMedium(mp);
	}

	return(HA_OK);
}

/*
 * Create new UDP/IP multicast heartbeat object 
 * pass in name of interface, multicast address, port, multicast
 * ttl, and multicast loopback value as parameters.
 * This should get called from hb_dev_parse().
 */
static struct hb_media *
mcast_new(const char * intf, const char *mcast, u_short port,
		    u_char ttl, u_char loop)
{
	struct mcast_private*	mcp;
	struct hb_media *	ret;

	/* create new mcast_private struct...hmmm...who frees it? */
	mcp = new_mcast_private(intf, mcast, port, ttl, loop);
	if (mcp == NULL) {
		PILCallLog(LOG, PIL_WARN, "Error creating mcast_private(%s, %s, %d, %d, %d)",
			 intf, mcast, port, ttl, loop);
		return(NULL);
	}
	ret = (struct hb_media*) MALLOC(sizeof(struct hb_media));
	if (ret != NULL) {
		char * name;
		ret->pd = (void*)mcp;
		name = STRDUP(intf);
		if (name != NULL) {
			ret->name = name;
		}
		else {
			FREE(ret);
			ret = NULL;
		}

	}
	if(ret == NULL) {
		FREE(mcp->interface);
		FREE(mcp);
	}
	return(ret);
}

/*
 *	Open UDP/IP multicast heartbeat interface
 */
static int
mcast_open(struct hb_media* hbm)
{
	struct mcast_private * mcp;

	MCASTASSERT(hbm);
	mcp = (struct mcast_private *) hbm->pd;

	if ((mcp->wsocket = mcast_make_send_sock(hbm)) < 0) {
		return(HA_FAIL);
	}
	if ((mcp->rsocket = mcast_make_receive_sock(hbm)) < 0) {
		mcast_close(hbm);
		return(HA_FAIL);
	}

	PILCallLog(LOG, PIL_INFO, "UDP multicast heartbeat started for group %s "
		"port %d interface %s (ttl=%d loop=%d)" , inet_ntoa(mcp->mcast),
		mcp->port, mcp->interface, mcp->ttl, mcp->loop);

	return(HA_OK);
}

/*
 *	Close UDP/IP multicast heartbeat interface
 */
static int
mcast_close(struct hb_media* hbm)
{
	struct mcast_private * mcp;
	int	rc = HA_OK;

	MCASTASSERT(hbm);
	mcp = (struct mcast_private *) hbm->pd;

	if (mcp->rsocket >= 0) {
		if (close(mcp->rsocket) < 0) {
			rc = HA_FAIL;
		}
	}
	if (mcp->wsocket >= 0) {
		if (close(mcp->wsocket) < 0) {
			rc = HA_FAIL;
		}
	}
	return(rc);
}

/*
 * Receive a heartbeat multicast packet from UDP interface
 */

static void *
mcast_read(struct hb_media* hbm, int *lenp)
{
	struct mcast_private *	mcp;
	char			buf[MAXLINE];
	int			addr_len = sizeof(struct sockaddr);
   	struct sockaddr_in	their_addr; /* connector's addr information */
	int	numbytes;
	void	*pkt;

	MCASTASSERT(hbm);
	mcp = (struct mcast_private *) hbm->pd;
	
	if ((numbytes=recvfrom(mcp->rsocket, buf, MAXLINE-1, 0
			       ,(struct sockaddr *)&their_addr, &addr_len)) < 0) {
		if (errno != EINTR) {
			PILCallLog(LOG, PIL_CRIT, "Error receiving from socket: %s"
			    ,	strerror(errno));
		}
		return NULL;
	}
	/* Avoid possible buffer overruns */
	buf[numbytes] = EOS;
	
	if (Debug >= PKTTRACE) {
		PILCallLog(LOG, PIL_DEBUG, "got %d byte packet from %s"
		    ,	numbytes, inet_ntoa(their_addr.sin_addr));
	}
	if (Debug >= PKTCONTTRACE && numbytes > 0) {
		PILCallLog(LOG, PIL_DEBUG, "%s", buf);
	}
	
	pkt = ha_malloc(numbytes + 1);
	if(!pkt){
		PILCallLog(LOG, PIL_CRIT
		    ,	"Error in allocating memory");
		return(NULL);
	}
	
	memcpy(pkt, buf, numbytes + 1);
	*lenp = numbytes + 1 ;

	return(pkt);
	
	
}

/*
 * Send a heartbeat packet over multicast UDP/IP interface
 */

static int
mcast_write(struct hb_media* hbm, void *pkt, int len)
{
	struct mcast_private *	mcp;
	int			rc;
	
	MCASTASSERT(hbm);
	mcp = (struct mcast_private *) hbm->pd;

	if ((rc=sendto(mcp->wsocket, pkt, len, 0
	,	(struct sockaddr *)&mcp->addr
	,	sizeof(struct sockaddr))) != len) {
		PILCallLog(LOG, PIL_CRIT, "Unable to send mcast packet [%d]: %s"
		,	rc, strerror(errno));
		return(HA_FAIL);
	}
	
	if (Debug >= PKTTRACE) {
		PILCallLog(LOG, PIL_DEBUG, "sent %d bytes to %s"
		    ,	rc, inet_ntoa(mcp->addr.sin_addr));
   	}
	if (Debug >= PKTCONTTRACE) {
		PILCallLog(LOG, PIL_DEBUG, "%s", (const char *)pkt);
   	}
	return(HA_OK);


  return(HA_OK);
  
}