ping_group.c

在LINUX下实现HA的源代码

 */

static void *
ping_group_read(struct hb_media* mp, int *lenp)
{
	
	ping_group_private_t *	ei;
	union {
		char		cbuf[MAXLINE+ICMP_HDR_SZ];
		struct ip	ip;
	}buf;
	const char *		bufmax = ((char *)&buf)+sizeof(buf);
	char *			msgstart;
	socklen_t		addr_len = sizeof(struct sockaddr);
   	struct sockaddr_in	their_addr; /* connector's addr information */
	struct ip *		ip;
	struct icmp		icp;
	int			numbytes;
	int			hlen;
	int 			seq;
	size_t			slotn;
	ping_group_node_t	*node;
	struct ha_msg		*msg = NULL;
	const char 		*comment;
	char			*pkt;
	int			pktlen;
	
	PINGGROUPASSERT(mp);
	ei = (ping_group_private_t *) mp->pd;

ReRead:        /* We recv lots of packets that aren't ours */

	if ((numbytes=recvfrom(ei->sock, (void *) &buf.cbuf
	,	sizeof(buf.cbuf)-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 potential buffer overruns */
	buf.cbuf[numbytes] = EOS;

	/* Check the IP header */
	ip = &buf.ip;
	hlen = ip->ip_hl * 4;

	if (numbytes < hlen + ICMP_MINLEN) {
		PILCallLog(LOG, PIL_WARN, "ping packet too short (%d bytes) from %s"
		,	numbytes
		,	inet_ntoa(*(struct in_addr *)
		&		their_addr.sin_addr.s_addr));
		return(NULL);
	}

	/* Now the ICMP part */	/* (there may be a better way...) */
	memcpy(&icp, (buf.cbuf + hlen), sizeof(icp));

	if (icp.icmp_type != ICMP_ECHOREPLY || icp.icmp_id != ei->ident) {
		goto ReRead;	/* Not ours */
	}
	seq = ntohs(icp.icmp_seq);

	if (DEBUGPKT) {
		PILCallLog(LOG, PIL_DEBUG, "got %d byte packet from %s"
		,	numbytes, inet_ntoa(their_addr.sin_addr));
	}
	msgstart = (buf.cbuf + hlen + ICMP_HDR_SZ);

	if (DEBUGPKTCONT && numbytes > 0) {
		PILCallLog(LOG, PIL_DEBUG, "%s", msgstart);
	}


	for(node = ei->node; node; node = node->next) {
		if(!memcmp(&(their_addr.sin_addr), &(node->addr.sin_addr),
					sizeof(struct in_addr))) {
			goto ReRead;    /* Not ours */
		}
	}
	if(!node) {
		return(NULL);
	}

	msg = wirefmt2msg(msgstart, bufmax - msgstart);
	if(msg == NULL) {
		errno = EINVAL;
		return(NULL);
	}
	comment = ha_msg_value(msg, F_COMMENT);
	if(comment == NULL || strcmp(comment, PIL_PLUGIN_S)) {
		ha_msg_del(msg);
		errno = EINVAL;
		return(NULL);
	}

	slotn = seq % NSLOT;
	if(ei->slot[slotn] == seq) {
		/* Duplicate within window */
		ha_msg_del(msg);
		goto ReRead;	/* Not ours */
	}

	ei->slot[slotn] = seq;
	
	pktlen = numbytes - hlen - ICMP_HDR_SZ;
	if (NULL == (pkt = ha_malloc(pktlen + 1))) {
		ha_msg_del(msg);
		errno = ENOMEM;
		return NULL;
	}
	pkt[pktlen] = 0;
	
	memcpy(pkt, buf.cbuf + hlen + ICMP_HDR_SZ, pktlen);	
	*lenp = pktlen + 1;
	
	ha_msg_del(msg);
	return(pkt);
}

/*
 * Send a heartbeat packet over broadcast UDP/IP interface
 *
 * The peculiar thing here is that we don't send the packet we're given at all
 *
 * Instead, we send out the packet we want to hear back from them, just
 * as though we were they ;-)  That's what comes of having such a dumb
 * device as a "member" of our cluster...
 *
 * We ignore packets we're given to write that aren't "status" packets.
 *
 */
static int
ping_group_write(struct hb_media* mp, void *p, int len)
{
	
	ping_group_private_t *	ei;
	int			rc;
	char*			pkt;
	union{
		char*			buf;
		struct icmp		ipkt;
	}*icmp_pkt;
	size_t			size;
	struct icmp *		icp;
	size_t			pktsize;
	const char *		type;
	const char *		ts;
	struct ha_msg *		nmsg;
	ping_group_node_t *	node;
	struct ha_msg*		msg;

	PINGGROUPASSERT(mp);

	
	if ((msg = wirefmt2msg(p, len)) == NULL) {
		PILCallLog(LOG, PIL_CRIT, "ping_write(): cannot convert wirefmt to msg");
		return(HA_FAIL);
	}



	ei = (ping_group_private_t *) mp->pd;
	type = ha_msg_value(msg, F_TYPE);

	if (type == NULL || strcmp(type, T_STATUS) != 0 
	|| ((ts = ha_msg_value(msg, F_TIME)) == NULL)) {
		ha_msg_del(msg);
		return HA_OK;
	}

	/*
	 * We populate the following fields in the packet we create:
	 *
	 * F_TYPE:	T_NS_STATUS
	 * F_STATUS:	ping
	 * F_COMMENT:	ping_group
	 * F_ORIG:	destination name
	 * F_TIME:	local timestamp (from "msg")
	 * F_AUTH:	added by add_msg_auth()
	 */
	if ((nmsg = ha_msg_new(5)) == NULL) {
		PILCallLog(LOG, PIL_CRIT, "cannot create new message");
		return(HA_FAIL);
	}

	if (ha_msg_add(nmsg, F_TYPE, T_NS_STATUS) != HA_OK
	||	ha_msg_add(nmsg, F_STATUS, PINGSTATUS) != HA_OK
	||	ha_msg_add(nmsg, F_COMMENT, PIL_PLUGIN_S) != HA_OK
	||	ha_msg_add(nmsg, F_ORIG, mp->name) != HA_OK
	||	ha_msg_add(nmsg, F_TIME, ts) != HA_OK) {
		ha_msg_del(nmsg); nmsg = NULL;
		PILCallLog(LOG, PIL_CRIT, "cannot add fields to message");
		ha_msg_del(msg);
		return HA_FAIL;
	}

	if (add_msg_auth(nmsg) != HA_OK) {
		PILCallLog(LOG, PIL_CRIT, "cannot add auth field to message");
		ha_msg_del(nmsg); nmsg = NULL;
		ha_msg_del(msg);
		return HA_FAIL;
	}

	if ((pkt = msg2wirefmt(nmsg, &size)) == NULL)  {
		PILCallLog(LOG, PIL_CRIT, "cannot convert message to string");
		ha_msg_del(msg);
		return HA_FAIL;
	}
	ha_msg_del(nmsg); nmsg = NULL;


	pktsize = size + ICMP_HDR_SZ;

	if ((icmp_pkt = MALLOC(pktsize)) == NULL) {
		PILCallLog(LOG, PIL_CRIT, "out of memory");
		ha_free(pkt);
		ha_msg_del(msg);
		return HA_FAIL;
	}

	icp = &(icmp_pkt->ipkt);
	icp->icmp_type = ICMP_ECHO;
	icp->icmp_code = 0;
	icp->icmp_cksum = 0;
	icp->icmp_seq = htons(ei->iseq);
	icp->icmp_id = ei->ident;	/* Only used by us */
	++ei->iseq;

	memcpy(icp->icmp_data, pkt, size);
	ha_free(pkt); pkt = NULL;

	/* Compute the ICMP checksum */
	icp->icmp_cksum = in_cksum((u_short *)icp, pktsize);

	for(node = ei->node; node; node = node->next) {
		if ((rc=sendto(ei->sock, (void *) icmp_pkt, pktsize, 0
		,	(struct sockaddr *)&node->addr
		,	sizeof(struct sockaddr))) != (ssize_t)pktsize) {
			PILCallLog(LOG, PIL_CRIT, "Error sending packet: %s"
			,	strerror(errno));
			FREE(icmp_pkt);
			ha_msg_del(msg);
			return(HA_FAIL);
		}

		if (DEBUGPKT) {
			PILCallLog(LOG, PIL_DEBUG, "sent %d bytes to %s"
			,	rc, inet_ntoa(node->addr.sin_addr));
   		}

		cl_shortsleep();
	}

	if (DEBUGPKTCONT) {
		PILCallLog(LOG, PIL_DEBUG, "%s", (const char*)pkt);
   	}

	FREE(icmp_pkt);
	ha_msg_del(msg);
	return HA_OK;
}

/*
 *	Open ping socket.
 */

static int
ping_group_open(struct hb_media* mp)
{
	ping_group_private_t * ei;
	int sockfd;
	struct protoent *proto;

	PINGGROUPASSERT(mp);
	ei = (ping_group_private_t *) mp->pd;


	if ((proto = getprotobyname("icmp")) == NULL) {
		PILCallLog(LOG, PIL_CRIT, "protocol ICMP is unknown: %s", strerror(errno));
		return HA_FAIL;
	}
	if ((sockfd = socket(AF_INET, SOCK_RAW, proto->p_proto)) < 0) {
		PILCallLog(LOG, PIL_CRIT, "Can't open RAW socket.: %s", strerror(errno));
		return HA_FAIL;
    	}

	if (fcntl(sockfd, F_SETFD, FD_CLOEXEC)) {
		PILCallLog(LOG, PIL_CRIT, "Error setting the close-on-exec flag: %s"
		,	strerror(errno));
	}
	ei->sock = sockfd;

	PILCallLog(LOG, PIL_INFO, "ping group heartbeat started.");
	return HA_OK;
}

/*
 * in_cksum --
 *	Checksum routine for Internet Protocol family headers (C Version)
 *	This function taken from Mike Muuss' ping program.
 */
static int
in_cksum (u_short *addr, size_t len)
{
	size_t		nleft = len;
	u_short *	w = addr;
	int		sum = 0;
	u_short		answer = 0;

	/*
	 * The IP checksum algorithm is simple: using a 32 bit accumulator (sum)
	 * add sequential 16 bit words to it, and at the end, folding back all
	 * the carry bits from the top 16 bits into the lower 16 bits.
	 */
	while (nleft > 1) {
		sum += *w++;
		nleft -= 2;
	}

	/* Mop up an odd byte, if necessary */
	if (nleft == 1) {
		sum += *(u_char*)w;
	}

	/* Add back carry bits from top 16 bits to low 16 bits */

	sum = (sum >> 16) + (sum & 0xffff);	/* add hi 16 to low 16 */
	sum += (sum >> 16);			/* add carry */
	answer = ~sum;				/* truncate to 16 bits */

	return answer;
}


/* 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
ping_group_parse(const char *line)
{
	char		tmp[MAXLINE];
	size_t		len;
	size_t		nhost = 0;
	struct hb_media *media;

	/* Skip over white space, then grab the name */
	line += strspn(line, WHITESPACE);
	len = strcspn(line, WHITESPACE);
	strncpy(tmp, line, len);
	line += len;
	*(tmp+len) = EOS;

	if(*tmp == EOS) {
		return(HA_FAIL);
	}

	media = ping_group_new(tmp);
	if (!media) {
		return(HA_FAIL);
	}

	while(1) {
		/* Skip over white space, then grab the host */
		line += strspn(line, WHITESPACE);
		len = strcspn(line, WHITESPACE);
		strncpy(tmp, line, len);
		line += len;
		*(tmp+len) = EOS;

		if(*tmp == EOS) {
			break;
		}

		if(ping_group_add_node(media, tmp) < 0) {
			ping_group_destroy(media);
			return(HA_FAIL);
		}
		nhost++;
	}

	if(nhost == 0) {
		ping_group_destroy(media);
		return(HA_FAIL);
	}
		
	OurImports->RegisterNewMedium(media);

	return(HA_OK);
}