client_lib.c

在LINUX下实现HA的源代码

	/* Special case: client status (change) */

	if (p->cstatus_callback && strcasecmp(mtype, T_APICLISTAT) == 0) {
		p->cstatus_callback(ha_msg_value(msg, F_ORIG)
		,	ha_msg_value(msg, F_FROMID)
		,       ha_msg_value(msg, F_STATUS)
		,       p->client_private);
		return(1);
	}
	if (p->cstatus_callback && strcasecmp(mtype, T_RCSTATUS) == 0) {
		p->cstatus_callback(ha_msg_value(msg, F_ORIG)
		,	ha_msg_value(msg, F_CLIENTNAME)
		,       ha_msg_value(msg, F_CLIENTSTATUS)
		,       p->client_private);
		return(1);
	}

	/* The general case: Any other message type */

	if ((gcb = search_gen_callback(mtype, p)) != NULL) {
		gcb->cf(msg, gcb->pd);
		return 1;
	}
	return(0);
}
/*
 * Return the next message not handled by a callback.
 * Invoke callbacks for messages encountered along the way.
 */
static struct ha_msg *
read_msg_w_callbacks(ll_cluster_t* llc, int blocking)
{
	struct ha_msg*	msg = NULL;
	llc_private_t* pi;

	if (!ISOURS(llc)) {
		ha_api_log(LOG_ERR, "read_msg_w_callbacks: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*) llc->ll_cluster_private;

	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "read_msg_w_callbacks: Not signed on");
		return NULL;
	}
	do {
		if (msg) {
			ZAPMSG(msg);
		}
		msg = read_hb_msg(llc, blocking);

	}while (msg && CallbackCall(pi, msg));
	return(msg);
}
/*
 * Receive messages.  Activate callbacks.  Messages without callbacks
 * are ignored.  Potentially several messages could be acted on.
 */
static int
rcvmsg(ll_cluster_t* llc, int blocking)
{
	struct ha_msg*	msg = NULL;
	
	msg=read_msg_w_callbacks(llc, blocking);

	if (msg) {
		ZAPMSG(msg);
		return(1);
	}
	return(0);
}

/*
 * Initialize nodewalk. (mainly retrieve list of nodes)
 */
static int
init_nodewalk (ll_cluster_t* ci)
{
	llc_private_t*	pi;
	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "init_nodewalk: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)ci->ll_cluster_private;

	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	zap_nodelist(pi);

	return(get_nodelist(pi));
}

/*
 * Return the next node in the list, or NULL if none.
 */
static const char *
nextnode (ll_cluster_t* ci)
{
	llc_private_t*	pi;
	const char *	ret;

	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "nextnode: bad cinfo");
		return NULL;
	}
	pi = (llc_private_t*)ci->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return NULL;
	}
	if (pi->nextnode == NULL) {
		return(NULL);
	}
	ret = pi->nextnode->value;

	pi->nextnode = pi->nextnode->next;
	return(ret);
}
/*
 * Clean up after a nodewalk (throw away node list)
 */
static int
end_nodewalk(ll_cluster_t* ci)
{
	llc_private_t*	pi;
	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "end_nodewalk: bad cinfo");
		return HA_FAIL;
	}
	pi = ci->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	zap_nodelist(pi);
	return(HA_OK);
}

/*
 * Initialize interface walk. (mainly retrieve list of interfaces)
 */
static int
init_ifwalk (ll_cluster_t* ci, const char * host)
{
	llc_private_t*	pi;
	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "init_ifwalk: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)ci->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	zap_iflist(pi);
	return(get_iflist(pi, host));
}

/*
 * Return the next interface in the iflist, or NULL if none.
 */
static const char *
nextif (ll_cluster_t* ci)
{
	llc_private_t*	pi = ci->ll_cluster_private;
	const char *	ret;

	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "nextif: bad cinfo");
		return HA_FAIL;
	}
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	if (pi->nextif == NULL) {
		return(NULL);
	}
	ret = pi->nextif->value;

	pi->nextif = pi->nextif->next;
	return(ret);
}

/*
 * Clean up after a ifwalk (throw away interface list)
 */
static int
end_ifwalk(ll_cluster_t* ci)
{
	llc_private_t*	pi = ci->ll_cluster_private;
	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "end_ifwalk: bad cinfo");
		return HA_FAIL;
	}
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	zap_iflist(pi);
	return HA_OK;
}

/*
 * Return the input file descriptor associated with this object.
 */
static int
get_inputfd(ll_cluster_t* ci)
{
	llc_private_t* pi;
	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "get_inputfd: bad cinfo");
		return(-1);
	}
	pi = (llc_private_t*)ci->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return -1;
	}
	return pi->chan->ops->get_recv_select_fd(pi->chan);
}
/*
 * Return the IPC channel associated with this object.
 */
static IPC_Channel*
get_ipcchan(ll_cluster_t* ci)
{
	llc_private_t* pi;
	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "get_ipcchan: bad cinfo");
		return NULL;
	}
	pi = (llc_private_t*)ci->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return NULL;
	}
	return pi->chan;
}


/*
 * Return TRUE (1) if there is a message ready to read.
 */
static int
msgready(ll_cluster_t*ci )
{
	llc_private_t* pi;

	ClearLog();
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "msgready: bad cinfo");
		return 0;
	}
	pi = (llc_private_t*)ci->ll_cluster_private;

	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return 0;
	}
	if (pi->firstQdmsg) {
		return 1;
	}

	return pi->chan->ops->is_message_pending(pi->chan);
}

/*
 * Set message filter mode
 */
static int
setfmode(ll_cluster_t* lcl, unsigned mode)
{
	unsigned	filtermask;
	llc_private_t* pi;

	ClearLog();
	if (!ISOURS(lcl)) {
		ha_api_log(LOG_ERR, "setfmode: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)lcl->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	switch(mode) {

		case LLC_FILTER_DEFAULT:
			filtermask = DEFAULTREATMENT;
			break;
		case LLC_FILTER_PMODE:
			filtermask = (KEEPIT|DUPLICATE|DROPIT);
			break;
		case LLC_FILTER_ALLHB:
			filtermask = (KEEPIT|DUPLICATE|DROPIT|NOCHANGE);
			break;
		case LLC_FILTER_RAW:
			filtermask = ALLTREATMENTS;
			break;
		default:
			return(HA_FAIL);
	}
	return(hb_api_setfilter(lcl, filtermask));
	
}

/*
 * Send a message to the cluster.
 */

static int
sendclustermsg(ll_cluster_t* lcl, struct ha_msg* msg)
{
	llc_private_t* pi;
	ClearLog();
	if (!ISOURS(lcl)) {
		ha_api_log(LOG_ERR, "sendclustermsg: bad cinfo");
		return HA_FAIL;
	}

	pi = (llc_private_t*)lcl->ll_cluster_private;

	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}

	if (pi->iscasual) {
		ha_api_log(LOG_ERR, "sendclustermsg: casual client");
		return HA_FAIL;
	}

	return(msg2ipcchan(msg, pi->chan));
}

/*
 * Send a message to a specific node in the cluster.
 */
static int
sendnodemsg(ll_cluster_t* lcl, struct ha_msg* msg
,			const char * nodename)
{
	llc_private_t* pi;
	ClearLog();
	if (!ISOURS(lcl)) {
		ha_api_log(LOG_ERR, "sendnodemsg: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)lcl->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	if (pi->iscasual) {
		ha_api_log(LOG_ERR, "sendnodemsg: casual client");
		return HA_FAIL;
	}
	if (*nodename == EOS) {
		ha_api_log(LOG_ERR, "sendnodemsg: bad nodename");
		return HA_FAIL;
	}
	if (ha_msg_mod(msg, F_TO, nodename) != HA_OK) {
		ha_api_log(LOG_ERR, "sendnodemsg: cannot set F_TO field");
		return(HA_FAIL);
	}
	return(msg2ipcchan(msg, pi->chan));
}

/* Add order sequence number field */
STATIC  void
add_order_seq(llc_private_t* pi, struct ha_msg* msg)
{
        order_seq_t *	order_seq = &pi->order_seq_head;
        const char *	to_node;
        char		seq[32];

	to_node = ha_msg_value(msg, F_TO);
	if (to_node != NULL){
		for (order_seq = pi->order_seq_head.next; order_seq != NULL
		;	order_seq = order_seq->next){
			if (strcmp(order_seq->to_node, to_node) == 0)
				break;
		}
	}
	if (order_seq == NULL && to_node != NULL){
		order_seq = (order_seq_t *) ha_malloc(sizeof(order_seq_t));
		if (order_seq == NULL){
			ha_api_log(LOG_ERR
			,	"add_order_seq: order_seq_t malloc failed!");
			return;
            	}
		strncpy(order_seq->to_node, to_node, HOSTLENG);
		order_seq->seqno = 1;
		order_seq->next = pi->order_seq_head.next;
		pi->order_seq_head.next = order_seq;
        }
        sprintf(seq, "%lx", order_seq->seqno);
        order_seq->seqno++;
        ha_msg_mod(msg, F_ORDERSEQ, seq);
}

/*
 * Send an ordered message to the cluster.
 */

static int
send_ordered_clustermsg(ll_cluster_t* lcl, struct ha_msg* msg)
{
	llc_private_t* pi;
	ClearLog();
	if (!ISOURS(lcl)) {
		ha_api_log(LOG_ERR, "%s: bad cinfo", __FUNCTION__);
		return HA_FAIL;
	}

	pi = (llc_private_t*)lcl->ll_cluster_private;

	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}

	if (pi->iscasual) {
		ha_api_log(LOG_ERR, "%s: casual client", __FUNCTION__);
		return HA_FAIL;
	}

	add_order_seq(pi, msg);

	return(msg2ipcchan(msg, pi->chan));
}

static int
send_ordered_nodemsg(ll_cluster_t* lcl, struct ha_msg* msg
,			const char * nodename)
{
	llc_private_t* pi;
	ClearLog();
	if (!ISOURS(lcl)) {
		ha_api_log(LOG_ERR, "sendnodemsg: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)lcl->ll_cluster_private;
	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}
	if (pi->iscasual) {
		ha_api_log(LOG_ERR, "sendnodemsg: casual client");
		return HA_FAIL;
	}
	if (*nodename == EOS) {
		ha_api_log(LOG_ERR, "sendnodemsg: bad nodename");
		return HA_FAIL;
	}
	if (ha_msg_mod(msg, F_TO, nodename) != HA_OK) {
		ha_api_log(LOG_ERR, "sendnodemsg: cannot set F_TO field");
		return(HA_FAIL);
	}
	add_order_seq(pi, msg);
	return(msg2ipcchan(msg, pi->chan));
}

static char	APILogBuf[MAXLINE] = "";
size_t		BufLen = 0;

static void
ClearLog(void)
{
        memset(APILogBuf, 0, sizeof(APILogBuf));
	APILogBuf[0] = EOS;
	BufLen = 1;
}

static const char *
APIError(ll_cluster_t* lcl)
{
	return(APILogBuf);
}

static void
ha_api_log(int priority, const char * fmt, ...)
{
	size_t	len;
        va_list ap;
        char buf[MAXLINE];
 
        va_start(ap, fmt);
        vsnprintf(buf, sizeof(buf), fmt, ap);
        va_end(ap);
	len = strlen(buf);

	if ((BufLen + len) >= sizeof(APILogBuf)) {
		ClearLog();
	}
		
	if (APILogBuf[0] != EOS && APILogBuf[BufLen-1] != '\n') {
		strncat(APILogBuf, "\n", sizeof(APILogBuf)-BufLen-1);
		BufLen++;
	}

	strncat(APILogBuf, buf, sizeof(APILogBuf)-BufLen-1);
	BufLen += len;
}

static void
ha_api_perror(const char * fmt, ...)
{
	const char *	err;

	va_list ap;
	char buf[MAXLINE];

	err = strerror(errno);
	va_start(ap, fmt);
	vsnprintf(buf, MAXLINE, fmt, ap);
	va_end(ap);

	ha_api_log(LOG_ERR, "%s: %s", buf, err);

}

/*
 *	Our vector of member functions...
 */
static struct llc_ops heartbeat_ops = {
	hb_api_signon,		/* signon */
	hb_api_signoff,		/* signoff */
	hb_api_delete,		/* delete */
	set_msg_callback,	/* set_msg_callback */
	set_nstatus_callback,	/* set_nstatus_callback */
	set_ifstatus_callback,	/* set_ifstatus_callback */
	set_cstatus_callback,	/* set_cstatus_callback */
	init_nodewalk,		/* init_nodewalk */
	nextnode,		/* nextnode */
	end_nodewalk,		/* end_nodewalk */
	get_nodestatus,		/* node_status */
	get_nodetype,		/* node_type */
	init_ifwalk,		/* init_ifwalk */
	nextif,			/* nextif */
	end_ifwalk,		/* end_ifwalk */
	get_ifstatus,		/* if_status */
	get_clientstatus,	/* client_status */
	sendclustermsg,		/* sendclustermsg */
	sendnodemsg,		/* sendnodemsg */
	send_ordered_clustermsg,/* send_ordered_clustermsg */
	send_ordered_nodemsg,	/* send_ordered_nodemsg */
	get_inputfd,		/* inputfd */
	get_ipcchan,		/* ipcchan */
	msgready,		/* msgready */
	hb_api_setsignal,	/* setmsgsignal */
	rcvmsg,			/* rcvmsg */
	read_msg_w_callbacks,	/* readmsg */
	setfmode,		/* setfmode */
	get_parameter,		/* get_parameter */
	get_deadtime,		/* get_deadtime */
	get_keepalive,		/* get_keepalive */
	get_mynodeid,		/* get_mynodeid */
	get_logfacility,	/* suggested logging facility */
	get_resources,		/* Get current resource allocation */
	APIError,		/* errormsg */
};



/*
 * Create a new heartbeat API object
 */
static ll_cluster_t*
hb_cluster_new()
{
	ll_cluster_t*	ret;
	struct llc_private* hb;

	if ((hb = MALLOCT(struct llc_private)) == NULL) {
		return(NULL);
	}
	memset(hb, 0, sizeof(*hb));
	if ((ret = MALLOCT(ll_cluster_t)) == NULL) {
		ha_free(hb);
		hb = NULL;
		return(NULL);
	}
	memset(ret, 0, sizeof(*ret));

	hb->PrivateId = OurID;
	ret->ll_cluster_private = hb;
	ret->llc_ops = &heartbeat_ops;

	return ret;
}

/*
 * Create a new low-level cluster object of the specified type.
 */
ll_cluster_t*
ll_cluster_new(const char * llctype)
{
	if (strcmp(llctype, "heartbeat") == 0) {
		return hb_cluster_new();
	}
	return NULL;
}