lvm/rpc.mdcommd/mdmn_subr.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright 2009 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <thread.h>
#include "meta.h"
#include "mdmn_subr.h"

extern int mdmn_init_set(set_t setno, int todo);

uint_t mdmn_busy[MD_MAXSETS][MD_MN_NCLASSES];
mutex_t	mdmn_busy_mutex[MD_MAXSETS];
cond_t	mdmn_busy_cv[MD_MAXSETS];


/* the wakeup table for the initiator's side */
mdmn_wti_t mdmn_initiator_table[MD_MAXSETS][MD_MN_NCLASSES];

/* the wakeup table for the master */
mdmn_wtm_t mdmn_master_table[MD_MAXSETS][MD_MN_NCLASSES];

/* List of licensed ip addresses */
licensed_ip_t   licensed_nodes[NNODES];

/* speed up the search for licensed ip addresses */
md_mn_nodeid_t maxlicnodes = 0; /* 0 is not a valid node ID */

/*
 * Check if a given set/class combination is currently in use
 * If in use, returns TRUE
 * Otherwise returns FALSE
 *
 * Must be called with mdmn_busy_mutex held
 */
bool_t
mdmn_is_class_busy(set_t setno, md_mn_msgclass_t class)
{
	if (mdmn_busy[setno][class] & MDMN_BUSY) {
		return (TRUE);
	} else {
		return (FALSE);
	}
}

/*
 * Mark a given set/class combination as currently in use
 * If the class was already in use, returns FALSE
 * Otherwise returns TRUE
 *
 * So mdmn_mark_class_busy can be used like
 * if (mdmn_mark_class_busy(setno, class) == FALSE)
 * 	failure;
 * else
 *	success;
 *
 * Must be called with mdmn_busy_mutex held
 */
bool_t
mdmn_mark_class_busy(set_t setno, md_mn_msgclass_t class)
{
	if (mdmn_busy[setno][class] & MDMN_BUSY) {
		return (FALSE);
	} else {
		mdmn_busy[setno][class] |= MDMN_BUSY;
		commd_debug(MD_MMV_MISC, "busy: set=%d, class=%d\n",
		    setno, class);
		return (TRUE);
	}
}

/*
 * Mark a given set/class combination as currently available
 * Always succeeds, thus void.
 *
 * If this class is marked MDMN_SUSPEND_ALL, we are in the middle of
 * draining all classes of this set.
 * We have to mark class+1 as MDMN_SUSPEND_ALL too.
 * If class+2 wasn't busy, we proceed with class+2, and so on
 * If any class is busy, we return.
 * Then the drain process will be continued by the mdmn_mark_class_unbusy() of
 * that busy class
 */
void
mdmn_mark_class_unbusy(set_t setno, md_mn_msgclass_t class)
{
	commd_debug(MD_MMV_MISC, "unbusy: set=%d, class=%d\n", setno, class);
	mdmn_busy[setno][class] &= ~MDMN_BUSY;
	/* something changed, inform threads waiting for that */
	(void) cond_signal(&mdmn_busy_cv[setno]);

	if ((mdmn_busy[setno][class] & MDMN_SUSPEND_ALL) == 0) {
		return;
	}

	while (++class < MD_MN_NCLASSES) {
		commd_debug(MD_MMV_MISC,
		    "unbusy: suspending set=%d, class=%d\n", setno, class);
		if (mdmn_mark_class_suspended(setno, class, MDMN_SUSPEND_ALL)
		    == MDMNE_SET_NOT_DRAINED) {
			break;
		}
	}

}


/*
 * Check if a given set/class combination is locked.
 */
bool_t
mdmn_is_class_locked(set_t setno, md_mn_msgclass_t class)
{
	if (mdmn_busy[setno][class] & MDMN_LOCKED) {
		return (TRUE);
	} else {
		return (FALSE);
	}
}

/*
 * Mark a given set/class combination as locked.
 * No checking is done here, so routine can be void.
 * Locking a locked set/class is ok.
 *
 * Must be called with mdmn_busy_mutex held
 */
void
mdmn_mark_class_locked(set_t setno, md_mn_msgclass_t class)
{
	mdmn_busy[setno][class] |= MDMN_LOCKED;
}

/*
 * Mark a given set/class combination as unlocked.
 * No checking is done here, so routine can be void.
 * Unlocking a unlocked set/class is ok.
 *
 * Must be called with mdmn_busy_mutex held
 */
void
mdmn_mark_class_unlocked(set_t setno, md_mn_msgclass_t class)
{
	mdmn_busy[setno][class] &= ~MDMN_LOCKED;
}

/*
 * Suspend a set/class combination
 *
 * If called during draining all classes of a set susptype is MDMN_SUSPEND_ALL.
 * If only one class is about to be drained susptype is MDMN_SUSPEND_1.
 *
 * Returns:
 *	MDMNE_ACK if there are no outstanding messages
 *	MDMNE_SET_NOT_DRAINED otherwise
 *
 * Must be called with mdmn_busy_mutex held for this set.
 */
int
mdmn_mark_class_suspended(set_t setno, md_mn_msgclass_t class, uint_t susptype)
{
	/*
	 * We use the mdmn_busy array to mark this set is suspended.
	 */
	mdmn_busy[setno][class] |= susptype;

	/*
	 * If there are outstanding messages for this set/class we
	 * return MDMNE_SET_NOT_DRAINED, otherwise we return MDMNE_ACK
	 */
	if (mdmn_is_class_busy(setno, class) == TRUE) {
		return (MDMNE_SET_NOT_DRAINED);
	}
	return (MDMNE_ACK);
}

/*
 * Resume operation for a set/class combination after it was
 * previously suspended
 *
 * If called from mdmn_comm_resume_svc_1 to resume _one_ specific class
 * then susptype will be MDMN_SUSPEND_1
 * Otherwise to resume all classes of one set,
 * then susptype equals (MDMN_SUSPEND_ALL | MDMN_SUSPEND_1)
 *
 * Always succeeds, thus void.
 *
 * Must be called with mdmn_busy_mutex held for this set.
 */
void
mdmn_mark_class_resumed(set_t setno, md_mn_msgclass_t class, uint_t susptype)
{
	/* simply the reverse operation to mdmn_mark_set_drained() */
	mdmn_busy[setno][class] &= ~susptype;
}

/*
 * Check if a drain command was issued for this set/class combination.
 *
 * Must be called with mdmn_busy_mutex held for this set.
 */
bool_t
mdmn_is_class_suspended(set_t setno, md_mn_msgclass_t class)
{
	if (mdmn_busy[setno][class] & (MDMN_SUSPEND_ALL | MDMN_SUSPEND_1)) {
		return (TRUE);
	} else {
		return (FALSE);
	}
}

/*
 * Put a result into the wakeup table for the master
 * It's ensured that the msg id from the master_table entry and from
 * result are matching
 */
void
mdmn_set_master_table_res(set_t setno, md_mn_msgclass_t class,
				md_mn_result_t  *res)
{
	mdmn_master_table[setno][class].wtm_result = res;
}
void
mdmn_set_master_table_id(set_t setno, md_mn_msgclass_t class, md_mn_msgid_t *id)
{
	MSGID_COPY(id, &(mdmn_master_table[setno][class].wtm_id));
}

void
mdmn_set_master_table_addr(set_t setno, md_mn_msgclass_t class,
    md_mn_nodeid_t nid)
{
	mdmn_master_table[setno][class].wtm_addr = nid;
}


md_mn_result_t *
mdmn_get_master_table_res(set_t setno, md_mn_msgclass_t class)
{
	return (mdmn_master_table[setno][class].wtm_result);
}

void
mdmn_get_master_table_id(set_t setno, md_mn_msgclass_t class, md_mn_msgid_t *id)
{
	MSGID_COPY(&(mdmn_master_table[setno][class].wtm_id), id);
}

cond_t *
mdmn_get_master_table_cv(set_t setno, md_mn_msgclass_t class)
{
	return (&(mdmn_master_table[setno][class].wtm_cv));
}

mutex_t *
mdmn_get_master_table_mx(set_t setno, md_mn_msgclass_t class)
{
	return (&(mdmn_master_table[setno][class].wtm_mx));
}

md_mn_nodeid_t
mdmn_get_master_table_addr(set_t setno, md_mn_msgclass_t class)
{
	return (mdmn_master_table[setno][class].wtm_addr);
}


/* here come the functions dealing with the wakeup table for the initiators */


void
mdmn_register_initiator_table(set_t setno, md_mn_msgclass_t class,
    md_mn_msg_t *msg, SVCXPRT *transp)
{
	uint_t nnodes	= set_descriptor[setno]->sd_mn_numnodes;
	time_t timeout	= mdmn_get_timeout(msg->msg_type);


	MSGID_COPY(&(msg->msg_msgid),
	    &(mdmn_initiator_table[setno][class].wti_id));
	mdmn_initiator_table[setno][class].wti_transp = transp;
	mdmn_initiator_table[setno][class].wti_args = (char *)msg;

	/*
	 * as the point in time where we want to be guaranteed to be woken up
	 * again, we chose the
	 * current time + nnodes times the timeout value for the message type
	 */
	mdmn_initiator_table[setno][class].wti_time =
	    time((time_t *)NULL) + (nnodes * timeout);
}

/*
 * If the set/class combination is currently busy, return MDMNE_CLASS_BUSY
 * Otherwise return MDMNE_ACK
 */
int
mdmn_check_initiator_table(set_t setno, md_mn_msgclass_t class)
{
	if ((mdmn_initiator_table[setno][class].wti_id.mid_nid == ~0u) &&
	    (mdmn_initiator_table[setno][class].wti_transp == (SVCXPRT *)NULL))
		return (MDMNE_ACK);
	return (MDMNE_CLASS_BUSY);
}

/*
 * Remove an entry from the initiator table entirely,
 * This must be done with mutex held.
 */
void
mdmn_unregister_initiator_table(set_t setno, md_mn_msgclass_t class)
{
	mdmn_initiator_table[setno][class].wti_id.mid_nid = ~0u;
	mdmn_initiator_table[setno][class].wti_id.mid_time = 0LL;
	mdmn_initiator_table[setno][class].wti_transp = (SVCXPRT *)NULL;
	mdmn_initiator_table[setno][class].wti_args = (char *)0;
	mdmn_initiator_table[setno][class].wti_time = (time_t)0;
}

void
mdmn_get_initiator_table_id(set_t setno, md_mn_msgclass_t class,
				md_mn_msgid_t *mid)
{
	MSGID_COPY(&(mdmn_initiator_table[setno][class].wti_id), mid);
}

SVCXPRT *
mdmn_get_initiator_table_transp(set_t setno, md_mn_msgclass_t class)
{
	return (mdmn_initiator_table[setno][class].wti_transp);
}

char *
mdmn_get_initiator_table_args(set_t setno, md_mn_msgclass_t class)
{
	return (mdmn_initiator_table[setno][class].wti_args);
}

mutex_t *
mdmn_get_initiator_table_mx(set_t setno, md_mn_msgclass_t class)
{
	return (&(mdmn_initiator_table[setno][class].wti_mx));
}

time_t
mdmn_get_initiator_table_time(set_t setno, md_mn_msgclass_t class)
{
	return (mdmn_initiator_table[setno][class].wti_time);
}

extern uint_t	md_commd_global_verb;	/* global bitmask for debug classes */
extern FILE	*commdout;		/* debug output file for the commd */
extern hrtime_t __savetime;


/*
 * Print debug messages to the terminal or to syslog
 * commd_debug(MD_MMV_SYSLOG,....) is always printed (and always via syslog),
 * even if md_commd_global_verb is zero.
 *
 * Otherwise the correct bit must be set in the bitmask md_commd_global_verb
 */
void
commd_debug(uint_t debug_class, const char *message, ...)
{
	va_list ap;

	/* Is this a message for syslog? */
	if (debug_class == MD_MMV_SYSLOG) {

		va_start(ap, message);
		(void) vsyslog(LOG_WARNING, message, ap);
		va_end(ap);
	} else {
		/* Is this debug_class set in the global verbosity state?  */
		if ((md_commd_global_verb & debug_class) == 0) {
			return;
		}
		/* Is our output file already functioning? */
		if (commdout == NULL) {
			return;
		}
		/* Are timestamps activated ? */
		if (md_commd_global_verb & MD_MMV_TIMESTAMP) {
			/* print time since last TRESET in usecs */
			(void) fprintf(commdout, "[%s]",
			    meta_print_hrtime(gethrtime() - __savetime));
		}
		/* Now print the real message */
		va_start(ap, message);
		(void) vfprintf(commdout, message, ap);
		va_end(ap);
	}
}


void
dump_hex(uint_t debug_class, unsigned int *x, int cnt)
{
	cnt /= sizeof (unsigned int);
	while (cnt--) {
		commd_debug(debug_class, "0x%8x ", *x++);
		if (cnt % 4)
			continue;
		commd_debug(debug_class, "\n");
	}
	commd_debug(debug_class, "\n");
}

/* debug output: dump a message */
void
dump_msg(uint_t dbc, char *prefix, md_mn_msg_t *msg)
{
	commd_debug(dbc, "%s &msg	= 0x%x\n", prefix, (int)msg);
	commd_debug(dbc, "%s ID	= (%d, 0x%llx-%d)\n", prefix,
	    MSGID_ELEMS(msg->msg_msgid));
	commd_debug(dbc, "%s sender	= %d\n", prefix, msg->msg_sender);
	commd_debug(dbc, "%s flags	= 0x%x\n", prefix, msg->msg_flags);
	commd_debug(dbc, "%s setno	= %d\n", prefix, msg->msg_setno);
	commd_debug(dbc, "%s recipient  = %d\n", prefix, msg->msg_recipient);
	commd_debug(dbc, "%s type	= %d\n", prefix, msg->msg_type);
	commd_debug(dbc, "%s size	= %d\n", prefix, msg->msg_event_size);
	if (msg->msg_event_size) {
		commd_debug(dbc, "%s data	=\n", prefix);
		dump_hex(dbc, (unsigned int *)(void *)msg->msg_event_data,
		    msg->msg_event_size);
	}
}

/* debug output: dump a result structure */
void
dump_result(uint_t dbc, char *prefix, md_mn_result_t *res)
{
	commd_debug(dbc, "%s &res	= 0x%x\n", prefix, (int)res);
	commd_debug(dbc, "%s ID	= (%d, 0x%llx-%d)\n", prefix,
	    MSGID_ELEMS(res->mmr_msgid));
	commd_debug(dbc, "%s setno	= %d\n", prefix, res->mmr_setno);
	commd_debug(dbc, "%s type	= %d\n", prefix, res->mmr_msgtype);
	commd_debug(dbc, "%s flags	= 0x%x\n", prefix, res->mmr_flags);
	commd_debug(dbc, "%s comm_state= %d\n", prefix, res->mmr_comm_state);
	commd_debug(dbc, "%s exitval	= %d\n", prefix, res->mmr_exitval);
	commd_debug(dbc, "%s out_size	= %d\n", prefix, res->mmr_out_size);
	if (res->mmr_out_size)
		commd_debug(dbc, "%s out	= %s\n", prefix, res->mmr_out);
	commd_debug(dbc, "%s err_size	= %d\n", prefix, res->mmr_err_size);
	if (res->mmr_err_size)
		commd_debug(dbc, "%s err	= %s\n", prefix, res->mmr_err);
}


/*
 * Here we find out, where to store or find the results for a given msg.
 *
 * Per set we have a pointer to a three dimensional array:
 * mct[set] -> mct_mce[NNODES][MD_MN_NCLASSES][MAX_SUBMESSAGES]
 * So, for every possible node and for every possible class we can store
 * MAX_SUBMESSAGES results.
 * the way to find the correct index is
 *	submessage +
 *	class * MAX_SUBMESSAGES +
 *	nodeid * MAX_SUBMESSAGES * MD_MN_NCLASSES.
 *
 * To find the correct address the index has to be multiplied
 * by the size of one entry.
 */
static md_mn_mce_t *
mdmn_get_mce_by_msg(md_mn_msg_t *msg)
{
	set_t	setno = msg->msg_setno;
	int	nodeid = msg->msg_msgid.mid_nid;
	int	submsg = msg->msg_msgid.mid_smid;
	int	mct_index;
	off_t	mct_offset;
	md_mn_msgclass_t class;

	if (mct[setno] != NULL) {
		if (mdmn_init_set(setno, MDMN_SET_MCT) != 0) {
			return ((md_mn_mce_t *)MDMN_MCT_ERROR);
		}
	}

	if (submsg == 0) {
		class = mdmn_get_message_class(msg->msg_type);
	} else {
		class = msg->msg_msgid.mid_oclass;
	}

	mct_index = submsg + class * MAX_SUBMESSAGES +
	    nodeid * MAX_SUBMESSAGES * MD_MN_NCLASSES;

	mct_offset = mct_index * sizeof (md_mn_mce_t);

	/* LINTED Pointer alignment */
	return ((md_mn_mce_t *)((caddr_t)(mct[setno]) + mct_offset));

	/*
	 * the lint clean version would be:
	 * return (&(mct[setno]->mct_mce[0][0][0]) + mct_index);
	 * :-)
	 */
}

/*
 * mdmn_mark_completion(msg, result, flag)
 * Stores the result of this message into the mmaped memory MCT[setno]
 * In case the same message comes along a second time we will know that
 * this message has already been processed and we can deliver the
 * results immediately.
 *
 * Before a message handler is called, the message in the MCT is flagged
 * as currently being processed (flag == MDMN_MCT_IN_PROGRESS).
 * This we need so we don't start a second handler for the same message.
 *
 * After a message handler is completed, this routine is called with
 * flag == MDMN_MCT_DONE and the appropriate result that we store in the MCT.
 * As MCT[setno] is memory mapped to disks, this information is persistent
 * even across a crash of the commd.
 * It doesn't have to be persistent across a reboot, though.
 *
 * Returns MDMN_MCT_DONE in case of success
 * Returns MDMN_MCT_ERROR in case of error creating the mct
 */
int
mdmn_mark_completion(md_mn_msg_t *msg, md_mn_result_t *result, uint_t flag)
{
	md_mn_mce_t	*mce;
	uint_t		offset_in_page;

	mce = mdmn_get_mce_by_msg(msg);
	if (mce == (md_mn_mce_t *)-1) {
		return (MDMN_MCT_ERROR);
	}
	offset_in_page = (uint_t)(caddr_t)mce % sysconf(_SC_PAGESIZE);

	(void) memset(mce, 0, sizeof (md_mn_mce_t));

	MSGID_COPY(&msg->msg_msgid, &mce->mce_result.mmr_msgid);
	if (flag == MDMN_MCT_IN_PROGRESS) {
		mce->mce_flags = MDMN_MCT_IN_PROGRESS;
		goto mmc_out;
	}

	/*
	 * In case the message flags indicate that the result should not be
	 * stored in the MCT, we return a MDMN_MCT_NOT_DONE,
	 * so the message will be processed at any rate,
	 * even if we process this message twice.
	 * this makes sense if the result of the message is a dynamic status
	 * and might have changed meanwhile.
	 */
	if (msg->msg_flags & MD_MSGF_NO_MCT) {
		return (MDMN_MCT_DONE);
	}

	/* This msg is no longer in progress */
	mce->mce_flags = MDMN_MCT_DONE;

	mce->mce_result.mmr_msgtype	    = result->mmr_msgtype;
	mce->mce_result.mmr_setno	    = result->mmr_setno;
	mce->mce_result.mmr_flags	    = result->mmr_flags;
	mce->mce_result.mmr_sender	    = result->mmr_sender;
	mce->mce_result.mmr_failing_node    = result->mmr_failing_node;
	mce->mce_result.mmr_comm_state	    = result->mmr_comm_state;
	mce->mce_result.mmr_exitval	    = result->mmr_exitval;

	/* if mmr_exitval is zero, we store stdout, otherwise stderr */
	if (result->mmr_exitval == 0) {
		if (result->mmr_out_size > 0) {
			(void) memcpy(mce->mce_data, result->mmr_out,
			    result->mmr_out_size);
			mce->mce_result.mmr_out_size = result->mmr_out_size;
		}
	} else {
		if (result->mmr_err_size > 0) {
			mce->mce_result.mmr_err_size = result->mmr_err_size;
			(void) memcpy(mce->mce_data, result->mmr_err,
			    result->mmr_err_size);
		}
	}

	dump_result(MD_MMV_PROC_S, "mdmn_mark_completion1", result);

mmc_out:
	/* now flush this entry to disk */
	(void) msync((caddr_t)mce - offset_in_page,
	    sizeof (md_mn_mce_t) + offset_in_page, MS_SYNC);
	return (MDMN_MCT_DONE);
}

/*
 * mdmn_check_completion(msg, resultp)
 * checks if msg has already been processed on this node, and if so copies
 * the stored result to resultp.
 *
 * returns MDMN_MCT_DONE and the result filled out acurately in case the
 *		msg has already been processed before
 * returns MDMN_MCT_NOT_DONE if the message has not been processed before
 * returns MDMN_MCT_IN_PROGRESS if the message is currently being processed
 *	This can only occur on a slave node.
 * return MDMN_MCT_ERROR in case of error creating the mct
 */
int
mdmn_check_completion(md_mn_msg_t *msg, md_mn_result_t *result)
{
	md_mn_mce_t	*mce;
	size_t		outsize;
	size_t		errsize;

	mce = mdmn_get_mce_by_msg(msg);
	if (mce == (md_mn_mce_t *)MDMN_MCT_ERROR) {
		return (MDMN_MCT_ERROR); /* what to do in that case ? */
	}
	if (MSGID_CMP(&(msg->msg_msgid), &(mce->mce_result.mmr_msgid))) {
		/* is the message completed, or in progress? */
		if (mce->mce_flags & MDMN_MCT_IN_PROGRESS) {
			return (MDMN_MCT_IN_PROGRESS);
		}
		/*
		 * See comment on MD_MSGF_NO_MCT above, if this flag is set
		 * for a message no result was stored and so the message has
		 * to be processed no matter if this is the 2nd time then.
		 */
		if (msg->msg_flags & MD_MSGF_NO_MCT) {
			return (MDMN_MCT_NOT_DONE);
		}

		/* Paranoia check: mce_flags must be MDMN_MCT_DONE here */
		if ((mce->mce_flags & MDMN_MCT_DONE) == 0) {
			commd_debug(MD_MMV_ALL,
			    "mdmn_check_completion: msg not done and not in "
			    "progress! ID = (%d, 0x%llx-%d)\n",
			    MSGID_ELEMS(msg->msg_msgid));
			return (MDMN_MCT_NOT_DONE);
		}
		/*
		 * Already processed.
		 * Copy saved results data;
		 * return only a pointer to any output.
		 */
		MSGID_COPY(&(mce->mce_result.mmr_msgid), &result->mmr_msgid);
		result->mmr_msgtype	    = mce->mce_result.mmr_msgtype;
		result->mmr_setno	    = mce->mce_result.mmr_setno;
		result->mmr_flags	    = mce->mce_result.mmr_flags;
		result->mmr_sender	    = mce->mce_result.mmr_sender;
		result->mmr_failing_node    = mce->mce_result.mmr_failing_node;
		result->mmr_comm_state	    = mce->mce_result.mmr_comm_state;
		result->mmr_exitval	    = mce->mce_result.mmr_exitval;
		result->mmr_err		    = NULL;
		result->mmr_out		    = NULL;
		outsize = result->mmr_out_size = mce->mce_result.mmr_out_size;
		errsize = result->mmr_err_size = mce->mce_result.mmr_err_size;
		/*
		 * if the exit val is zero only stdout was stored (if any)
		 * otherwise only stderr was stored (if any)
		 */
		if (result->mmr_exitval == 0) {
			if (outsize != 0) {
				result->mmr_out = Zalloc(outsize);
				(void) memcpy(result->mmr_out, mce->mce_data,
				    outsize);
			}
		} else {
			if (errsize != 0) {
				result->mmr_err = Zalloc(errsize);
				(void) memcpy(result->mmr_err, mce->mce_data,
				    errsize);
			}
		}
		commd_debug(MD_MMV_MISC,
		    "mdmn_check_completion: msg already processed \n");
		dump_result(MD_MMV_MISC, "mdmn_check_completion", result);
		return (MDMN_MCT_DONE);
	}
	commd_debug(MD_MMV_MISC,
	    "mdmn_check_completion: msg not yet processed\n");
	return (MDMN_MCT_NOT_DONE);
}


/*
 * check_license(rqstp, chknid)
 *
 * Is this RPC request sent from a licensed host?
 *
 * If chknid is non-zero, the caller of check_license() knows the ID of
 * the sender. Then we check just the one entry of licensed_nodes[]
 *
 * If chknid is zero, the sender is not known. In that case the sender must be
 * the local node.
 *
 * If the host is licensed, return TRUE, else return FALSE
 */
bool_t
check_license(struct svc_req *rqstp, md_mn_nodeid_t chknid)
{
	char		buf[INET6_ADDRSTRLEN];
	void		*caller = NULL;
	in_addr_t	caller_ipv4;
	in6_addr_t	caller_ipv6;
	struct sockaddr	*ca;


	ca = (struct sockaddr *)(void *)svc_getrpccaller(rqstp->rq_xprt)->buf;

	if (ca->sa_family == AF_INET) {
		caller_ipv4 =
		    ((struct sockaddr_in *)(void *)ca)->sin_addr.s_addr;
		caller = (void *)&caller_ipv4;

		if (chknid == 0) {
			/* check against local node */
			if (caller_ipv4 == htonl(INADDR_LOOPBACK)) {
				return (TRUE);

			}
		} else {
			/* check against one specific node */
			if ((caller_ipv4 == licensed_nodes[chknid].lip_ipv4) &&
			    (licensed_nodes[chknid].lip_family == AF_INET)) {
				return (TRUE);
			} else {
				commd_debug(MD_MMV_MISC,
				    "Bad attempt from %x ln[%d]=%x\n",
				    caller_ipv4, chknid,
				    licensed_nodes[chknid].lip_ipv4);
			}
		}
	} else if (ca->sa_family == AF_INET6) {
		caller_ipv6 = ((struct sockaddr_in6 *)(void *)ca)->sin6_addr;
		caller = (void *)&caller_ipv6;

		if (chknid == 0) {
			/* check against local node */
			if (IN6_IS_ADDR_LOOPBACK(&caller_ipv6)) {
				return (TRUE);

			}
		} else {
			/* check against one specific node */
			if (IN6_ARE_ADDR_EQUAL(&caller_ipv6,
			    &(licensed_nodes[chknid].lip_ipv6)) &&
			    (licensed_nodes[chknid].lip_family == AF_INET6)) {
				return (TRUE);
			}
		}
	}
	/* if  we are here, we were contacted by an unlicensed node */
	commd_debug(MD_MMV_SYSLOG,
	    "Bad attempt to contact rpc.mdcommd from %s\n",
	    caller ?
	    inet_ntop(ca->sa_family, caller, buf, INET6_ADDRSTRLEN) :
	    "unknown");

	return (FALSE);
}

/*
 * Add a node to the list of licensed nodes.
 *
 * Only IPv4 is currently supported.
 * for IPv6, we need to change md_mnnode_desc.
 */
void
add_license(md_mnnode_desc *node)
{
	md_mn_nodeid_t nid = node->nd_nodeid;
	char		buf[INET6_ADDRSTRLEN];

	/*
	 * If this node is not yet licensed, do it now.
	 * For now only IPv4 addresses are supported.
	 */
	commd_debug(MD_MMV_MISC, "add_lic(%s): ln[%d]=%s, lnc[%d]=%d\n",
	    node->nd_priv_ic, nid,
	    inet_ntop(AF_INET, (void *)&licensed_nodes[nid].lip_ipv4,
	    buf, INET6_ADDRSTRLEN), nid, licensed_nodes[nid].lip_cnt);

	if (licensed_nodes[nid].lip_ipv4 == (in_addr_t)0) {
		licensed_nodes[nid].lip_family = AF_INET; /* IPv4 */
		licensed_nodes[nid].lip_ipv4 = inet_addr(node->nd_priv_ic);
		/* keep track of the last entry for faster search */
		if (nid > maxlicnodes)
			maxlicnodes = nid;

	}
	/* in any case bump up the reference count */
	licensed_nodes[nid].lip_cnt++;
}

/*
 * lower the reference count for one node.
 * If that drops to zero, remove the node from the list of licensed nodes
 *
 * Only IPv4 is currently supported.
 * for IPv6, we need to change md_mnnode_desc.
 */
void
rem_license(md_mnnode_desc *node)
{
	md_mn_nodeid_t nid = node->nd_nodeid;
	char		buf[INET6_ADDRSTRLEN];

	commd_debug(MD_MMV_MISC, "rem_lic(%s): ln[%d]=%s, lnc[%d]=%d\n",
	    node->nd_priv_ic, nid,
	    inet_ntop(AF_INET, (void *)&licensed_nodes[nid].lip_ipv4, buf,
	    INET6_ADDRSTRLEN), nid, licensed_nodes[nid].lip_cnt);

	assert(licensed_nodes[nid].lip_cnt > 0);

	/*
	 * If this was the last reference to that node, it's license expires
	 * For now only IPv4 addresses are supported.
	 */
	if (--licensed_nodes[nid].lip_cnt == 0) {
		licensed_nodes[nid].lip_ipv4 = (in_addr_t)0;
	}
}